Apache Kafka-初体验Kafka(02)-Centos7下搭建单节点kafka_配置参数详解_基本命令实操

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: Apache Kafka-初体验Kafka(02)-Centos7下搭建单节点kafka_配置参数详解_基本命令实操


安装JDK

由于Kafka是用Scala语言开发的,运行在JVM上,因此在安装Kafka之前需要先安装JDK.

这里就不啰嗦了,戳这里—> : Java-CentoOS 7安装JDK8 (rpm格式 和 tar.gz格式)& 多JDK设置默认的Java 版本

[root@artisan ~]# java -version
java version "1.8.0_221"
Java(TM) SE Runtime Environment (build 1.8.0_221-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.221-b11, mixed mode)
[root@artisan ~]# 

安装zookeeper

kafka是基于zookeeper协调的分布式消息系统,所以zk也是必不可少的. kafka内置了一个zk, 不建议使用。

zookeeper 也是基于java开发的,所以也是需要依赖JDK的。


下载地址 -->: https://archive.apache.org/dist/zookeeper/

这里我们下载 3.4.14版本的, 事实上3.x版本的都可以。

[root@artisan zookeeper-3.4.14]# pwd
/usr/local/zookeeper-3.4.14
# copy配置文件
[root@artisan zookeeper-3.4.14]# cp conf/zoo_sample.cfg  conf/zoo.cfg
# 启动 zk 
[root@artisan zookeeper-3.4.14]# ./bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper-3.4.14/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
# 查看进程  QuorumPeerMain   --> zk的进程
[root@artisan zookeeper-3.4.14]# jps
3409 QuorumPeerMain
3425 Jps
# 查看状态
[root@artisan zookeeper-3.4.14]# ./bin/zkServer.sh  status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper-3.4.14/bin/../conf/zoo.cfg
Mode: standalone
[root@artisan zookeeper-3.4.14]# 
# 客户端连接
[root@artisan zookeeper-3.4.14]# ./bin/zkCli.sh 
Connecting to localhost:2181
2019-11-17 07:34:32,522 [myid:] - INFO  [main:Environment@100] - Client environment:zookeeper.version=3.4.14-4c25d480e66aadd371de8bd2fd8da255ac140bcf, built on 03/06/2019 16:18 GMT
2019-11-17 07:34:32,527 [myid:] - INFO  [main:Environment@100] - Client environment:host.name=192.168.18.130
2019-11-17 07:34:32,527 [myid:] - INFO  [main:Environment@100] - Client environment:java.version=1.8.0_221
2019-11-17 07:34:32,530 [myid:] - INFO  [main:Environment@100] - Client environment:java.vendor=Oracle Corporation
2019-11-17 07:34:32,530 [myid:] - INFO  [main:Environment@100] - Client environment:java.home=/usr/java/jdk1.8.0_221-amd64/jre
2019-11-17 07:34:32,530 [myid:] - INFO  [main:Environment@100] - Client environment:java.class.path=/usr/local/zookeeper-3.4.14/bin/../zookeeper-server/target/classes:/usr/local/zookeeper-3.4.14/bin/../build/classes:/usr/local/zookeeper-3.4.14/bin/../zookeeper-server/target/lib/*.jar:/usr/local/zookeeper-3.4.14/bin/../build/lib/*.jar:/usr/local/zookeeper-3.4.14/bin/../lib/slf4j-log4j12-1.7.25.jar:/usr/local/zookeeper-3.4.14/bin/../lib/slf4j-api-1.7.25.jar:/usr/local/zookeeper-3.4.14/bin/../lib/netty-3.10.6.Final.jar:/usr/local/zookeeper-3.4.14/bin/../lib/log4j-1.2.17.jar:/usr/local/zookeeper-3.4.14/bin/../lib/jline-0.9.94.jar:/usr/local/zookeeper-3.4.14/bin/../lib/audience-annotations-0.5.0.jar:/usr/local/zookeeper-3.4.14/bin/../zookeeper-3.4.14.jar:/usr/local/zookeeper-3.4.14/bin/../zookeeper-server/src/main/resources/lib/*.jar:/usr/local/zookeeper-3.4.14/bin/../conf:.:/usr/java/jdk1.8.0_221-amd64/jre/lib:/usr/java/jdk1.8.0_221-amd64/lib:/usr/java/jdk1.8.0_221-amd64/lib/tools.jar
2019-11-17 07:34:32,530 [myid:] - INFO  [main:Environment@100] - Client environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib
2019-11-17 07:34:32,530 [myid:] - INFO  [main:Environment@100] - Client environment:java.io.tmpdir=/tmp
2019-11-17 07:34:32,530 [myid:] - INFO  [main:Environment@100] - Client environment:java.compiler=<NA>
2019-11-17 07:34:32,530 [myid:] - INFO  [main:Environment@100] - Client environment:os.name=Linux
2019-11-17 07:34:32,531 [myid:] - INFO  [main:Environment@100] - Client environment:os.arch=amd64
2019-11-17 07:34:32,531 [myid:] - INFO  [main:Environment@100] - Client environment:os.version=3.10.0-123.el7.x86_64
2019-11-17 07:34:32,531 [myid:] - INFO  [main:Environment@100] - Client environment:user.name=root
2019-11-17 07:34:32,531 [myid:] - INFO  [main:Environment@100] - Client environment:user.home=/root
2019-11-17 07:34:32,531 [myid:] - INFO  [main:Environment@100] - Client environment:user.dir=/usr/local/zookeeper-3.4.14
2019-11-17 07:34:32,534 [myid:] - INFO  [main:ZooKeeper@442] - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@25f38edc
Welcome to ZooKeeper!
2019-11-17 07:34:32,624 [myid:] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@1025] - Opening socket connection to server localhost/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error)
JLine support is enabled
2019-11-17 07:34:32,940 [myid:] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@879] - Socket connection established to localhost/0:0:0:0:0:0:0:1:2181, initiating session
2019-11-17 07:34:33,066 [myid:] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@1299] - Session establishment complete on server localhost/0:0:0:0:0:0:0:1:2181, sessionid = 0x1000014c05c0000, negotiated timeout = 30000
WATCHER::
WatchedEvent state:SyncConnected type:None path:null
# 查看zk的根目录信息,默认只有 zookeeper 1个
[zk: localhost:2181(CONNECTED) 0] ls /
[zookeeper]
[zk: localhost:2181(CONNECTED) 1] 

安装kafka

下载解压

下载地址: https://kafka.apache.org/downloads

先说下 kafka版本的定义

kafka_2.11‐1.1.0 : 2.11 是 Scala的版本 ,1.1.0kafka的版本


20210102更新


配置hosts

启动kafka时会使用linux主机名关联的ip地址,所以需要把主机名和linux的ip映射配置到本地host里。

[root@artisan local]# cat /etc/hosts
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
192.168.18.130 artisan
[root@artisan local]# 

主要影响的点是这儿 server.properties

如果不配置的话, 那就是用 IP


启动kafka服务

启动脚本语法: kafka­server­start.sh [­daemon] server.properties

server.properties 的配置路径是一个强制的参数, ­daemon 表示以后台进程运行,否则ssh客户端退出后,就会停止服务

[root@artisan soft_artisan]# pwd
/usr/local/soft_artisan
[root@artisan soft_artisan]# tar -xvzf kafka_2.11-1.1.0.tgz  -C /usr/local/
[root@artisan soft_artisan]# cd /usr/local/kafka_2.11-1.1.0/
[root@artisan kafka_2.11-1.1.0]#  cd bin 
# 后台启动kafka
[root@artisan bin]# ./kafka-server-start.sh -daemon  ../config/server.properties
# 查看进程
[root@artisan bin]# jps   
3409 QuorumPeerMain
11923 Kafka
11942 Jps
[root@artisan bin]# 

查看zk下的节点信息

[root@artisan bin]# pwd
/usr/local/zookeeper-3.4.14/bin
[root@artisan bin]# 
[root@artisan bin]# ./zkCli.sh 
Connecting to localhost:2181
2019-11-17 10:05:52,083 [myid:] - INFO  [main:Environment@100] - Client environment:zookeeper.version=3.4.14
.....
.....
.....
# 根节点下,除了zookeeper以外,都是kafka创建的  
 #查看zk的根目录kafka相关节点
[zk: localhost:2181(CONNECTED) 0] ls /
[cluster, controller_epoch, controller, brokers, zookeeper, admin, isr_change_notification, consumers, log_dir_event_notification, latest_producer_id_block, config]
[zk: localhost:2181(CONNECTED) 1] ls /brokers
[ids, topics, seqid]
[zk: localhost:2181(CONNECTED) 2] ls /brokers/ids
[0
[zk: localhost:2181(CONNECTED) 3] 

server.properties核心配置详解

官方说明: https://kafka.apache.org/documentation/#configuration

我们挑几个常用的来说下吧

参数 默认值 说明
broker.id 0 每一个broker在集群中的唯一标识,非负数。当该服务器的IP地址发生改变时,broker.id没有变化,则不会影响consumers的消息情况
log.dirs /tmp/kafka-logs kafka数据的存放地址,多个地址的话用逗号分割,多个目录分布在不同磁盘上可以提高读写性能 /data/kafka-logs-1,/data/kafka-logs-2
listeners 9092 server接受客户端连接的端口
zookeeper.connect localhost:2181 zookeeper集群的地址,可以是多个,多个之间用逗号分割 hostname1:port1,hostname2:port2,hostname3:port3
log.retention.hours 168 每个日志文件删除之前保存的时间。默认数据保存时间对所有topic都一样。
min.insync.replicas 1 当producer设置acks为-1时,min.insync.replicas指定replicas的最小数目
delete.topic.enable false 是否允许删除主题

或者参考: apache kafka系列之server.properties配置文件参数说明


基本命令

官方指导: https://kafka.apache.org/quickstart

创建主题

主要是用kafka内置的 kafka-topics.sh 脚本 来操作消息 。

我们先来看下如何使用 该shell脚本吧

直接输入 kafka-topics.sh ,回车可以看到参数说明。

[root@artisan bin]# pwd
/usr/local/kafka_2.11-1.1.0/bin
[root@artisan bin]# ./kafka-topics.sh 
Create, delete, describe, or change a topic.
Option                                   Description                            
------                                   -----------                            
--alter                                  Alter the number of partitions,        
                                           replica assignment, and/or           
                                           configuration for the topic.         
--config <String: name=value>            A topic configuration override for the 
                                           topic being created or altered.The   
                                           following is a list of valid         
                                           configurations:                      
                                          cleanup.policy                        
                                          compression.type                      
                                          delete.retention.ms                   
                                          file.delete.delay.ms                  
                                          flush.messages                        
                                          flush.ms                              
                                          follower.replication.throttled.       
                                           replicas                             
                                          index.interval.bytes                  
                                          leader.replication.throttled.replicas 
                                          max.message.bytes                     
                                          message.format.version                
                                          message.timestamp.difference.max.ms   
                                          message.timestamp.type                
                                          min.cleanable.dirty.ratio             
                                          min.compaction.lag.ms                 
                                          min.insync.replicas                   
                                          preallocate                           
                                          retention.bytes                       
                                          retention.ms                          
                                          segment.bytes                         
                                          segment.index.bytes                   
                                          segment.jitter.ms                     
                                          segment.ms                            
                                          unclean.leader.election.enable        
                                         See the Kafka documentation for full   
                                           details on the topic configs.        
--create                                 Create a new topic.                    
--delete                                 Delete a topic                         
--delete-config <String: name>           A topic configuration override to be   
                                           removed for an existing topic (see   
                                           the list of configurations under the 
                                           --config option).                    
--describe                               List details for the given topics.     
--disable-rack-aware                     Disable rack aware replica assignment  
--force                                  Suppress console prompts               
--help                                   Print usage information.               
--if-exists                              if set when altering or deleting       
                                           topics, the action will only execute 
                                           if the topic exists                  
--if-not-exists                          if set when creating topics, the       
                                           action will only execute if the      
                                           topic does not already exist         
--list                                   List all available topics.             
--partitions <Integer: # of partitions>  The number of partitions for the topic 
                                           being created or altered (WARNING:   
                                           If partitions are increased for a    
                                           topic that has a key, the partition  
                                           logic or ordering of the messages    
                                           will be affected                     
--replica-assignment <String:            A list of manual partition-to-broker   
  broker_id_for_part1_replica1 :           assignments for the topic being      
  broker_id_for_part1_replica2 ,           created or altered.                  
  broker_id_for_part2_replica1 :                                                
  broker_id_for_part2_replica2 , ...>                                           
--replication-factor <Integer:           The replication factor for each        
  replication factor>                      partition in the topic being created.
--topic <String: topic>                  The topic to be create, alter or       
                                           describe. Can also accept a regular  
                                           expression except for --create option
--topics-with-overrides                  if set when describing topics, only    
                                           show topics that have overridden     
                                           configs                              
--unavailable-partitions                 if set when describing topics, only    
                                           show partitions whose leader is not  
                                           available                            
--under-replicated-partitions            if set when describing topics, only    
                                           show under replicated partitions     
--zookeeper <String: hosts>              REQUIRED: The connection string for    
                                           the zookeeper connection in the form 
                                           host:port. Multiple hosts can be     
                                           given to allow fail-over.            
[root@artisan bin]# 

那根据指导,创建个消息吧

--create                                 Create a new topic.                   
• 1

创建一个名字为“artisan”的Topic,这个topic只有一个partition,并且备份因子也设置为1: ./kafka-topics.sh --create --zookeeper 192.168.18.130:2181 --replication-factor 1 --partitions 1 --topic artisan

[root@artisan bin]# ./kafka-topics.sh --create --zookeeper 192.168.18.130:2181 --replication-factor 1 --partitions 1 --topic artisan
Created topic "artisan".
[root@artisan bin]# 

我们可以通过以下--list命令来查看kafka中目前存在的topic

[root@artisan bin]# ./kafka-topics.sh --list --zookeeper 192.168.18.130:2181
artisan
[root@artisan bin]# 

除了我们通过手工的方式创建Topic,当producer发布一个消息某个指定的Topic,但是这个Topic并不存在时,会自动创建

删除主题

# 删除
[root@artisan bin]# ./kafka-topics.sh --delete --topic artisan  --zookeeper 192.168.18.130:2181
Topic artisan is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
# 查看
[root@artisan bin]# ./kafka-topics.sh --list --zookeeper 192.168.18.130:2181
# 新建
[root@artisan bin]# ./kafka-topics.sh --create --zookeeper 192.168.18.130:2181 --replication-factor 1 --partitions 1 --topic artisan
Created topic "artisan".
[root@artisan bin]# 

发送消息

kafka自带了一个producer命令客户端,可以从本地文件中读取内容,或者我们也可以以命令行中直接输入内容,并将这些内容以消息的形式发送到kafka集群中。

在默认情况下,每一个行会被当做成一个独立的消息。

首先我们要运行发布消息的脚本,然后在命令中输入要发送的消息的内容

[root@artisan bin]# ./kafka-console-producer.sh --broker-list 192.168.18.130:9092 --topic artisan
>This is a message
>This is another message
>

消费消息

对于consumer,kafka同样也携带了一个命令行客户端,会将获取到内容在命令中进行输出,默认是消费最新的消息.

./kafka-console-consumer.sh --bootstrap-server 192.168.18.130:9092   --topic artisan

如果想要消费之前的消息可以通过--from-beginning参数指定,如下命令:

[root@artisan bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.18.130:9092  --from-beginning   --topic artisan 
This is a message
This is another message
This is artisan
this is anothhhh
this is artisan

如果你是通过不同的终端窗口来运行以上的命令,你将会看到在producer终端输入的内容,很快就会在consumer的终端窗口上显示出来。

以上所有的命令都有一些附加的选项;当我们不携带任何参数运行命令的时候,将会显示出这个命令的详细用法。


查看组名

[root@artisan bin]# ./kafka-consumer-groups.sh --bootstrap-server 192.168.18.130:9092 --list
Note: This will not show information about old Zookeeper-based consumers.
console-consumer-81551
console-consumer-72540
console-consumer-23504
testGroup
[root@artisan bin]# 

查看消费者的消费偏移量

[root@artisan bin]# ./kafka-consumer-groups.sh --bootstrap-server 192.168.18.130:9092 --describe --group testGroup
Note: This will not show information about old Zookeeper-based consumers.
Consumer group 'testGroup' has no active members.
TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
artisan         0          3               5               2               -               -               -
[root@artisan bin]# 

current-offset 和 log-end-offset还有 lag ,分别为当前消费偏移量,结束的偏移量(HW),落后消费的偏移量


消费多主题

先创建另外一个topic : xiaogongjiang

[root@artisan bin./kafka-topics.sh --create --zookeeper 192.168.18.130:2181 --replication-factor 1 --partitions 1 --topic xiaogongjiang
Created topic "xiaogongjiang".
[root@artisan bin]# 

开启两个生产者

[root@artisan bin]# ./kafka-console-producer.sh --broker-list 192.168.18.130:9092 --topic xiaogongjiang
>send from xiaogongjiang
>
[root@artisan bin]# ./kafka-console-producer.sh --broker-list 192.168.18.130:9092 --topic artisan
>send from artisan
>

消费多主题 如下

[root@artisan bin]# ./kafka-console-consumer.sh --bootstrap-server  192.168.18.130:9092 --whitelist "artisan|xiaogongjiang"
send from xiaogongjiang
send from artisan

单播消费

一条消息只能被某一个消费者消费的模式,类似queue模式,只需让所有消费者在同一个消费组里即可.

分别在两个客户端执行如下消费命令,然后往主题里发送消息,结果只有一个客户端能收到消息

这样,生产者跟消费组没关系,只要在消费的时候指定消费组即可

生产者

[root@artisan bin]# ./kafka-console-producer.sh --broker-list 192.168.18.130:9092 --topic artisan
>queue model test
>queue model the second message 

消费者1

[root@artisan bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.18.130:9092 --consumer-property group.id=artisanGroup --topic artisan

消费者2

[root@artisan bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.18.130:9092 --consumer-property group.id=artisanGroup --topic artisan
queue model test
queue model the second message 


多播消费

一条消息能被多个消费者消费的模式,类似publish-subscribe模式 费,针对Kafka同一条消息只能被同一个消费组下的某一个消费者消费的特性,要实现多播只要保证这些消费者属于不同的消费组即可。我们再增加一个消费者,该消费者属于 testGroup-2 消费组, 结果两个客户端都能收到消息.

生产者

[root@artisan bin]# ./kafka-console-producer.sh --broker-list 192.168.18.130:9092 --topic artisan
>messge artisan jajaja
>

消费者1 属于 anotherArtisanGroup消费组

[root@artisan bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.18.130:9092 --consumer-property group.id=anotherArtisanGroup --topic artisan
messge artisan jajaja

消费者1 属于 artisanGroup 消费组

[root@artisan bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.18.130:9092 --consumer-property group.id=artisanGroup --topic artisan
messge artisan jajaja


小结

到此为止,我们搭建了kafka的单节点环境,也演示了基本用法,接下来,我们来搭建一个3个节点的kafka集群吧。


相关文章
|
11天前
|
Linux 网络安全
Centos 防火墙端口控制命令
Centos 防火墙端口控制命令
26 3
|
1月前
|
消息中间件 缓存 关系型数据库
Flink CDC产品常见问题之upsert-kafka增加参数报错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
1月前
|
消息中间件 存储 大数据
Apache Kafka: 强大消息队列系统的介绍与使用
Apache Kafka: 强大消息队列系统的介绍与使用
|
1月前
|
SQL Apache HIVE
一文彻底掌握Apache Hudi的主键和分区配置
一文彻底掌握Apache Hudi的主键和分区配置
53 0
|
2月前
|
Java 程序员 API
Springboot-swagger配置(idea社区版2023.1.4+apache-maven-3.9.3-bin)
Springboot-swagger配置(idea社区版2023.1.4+apache-maven-3.9.3-bin)
55 1
|
1月前
|
安全 Linux Apache
Apache代理服务器搭建和配置
Apache代理服务器搭建和配置
|
30天前
|
消息中间件 Java Kafka
Kafka【环境搭建 01】kafka_2.12-2.6.0 单机版安装+参数配置及说明+添加到service服务+开机启动配置+验证+chkconfig配置说明(一篇入门kafka)
【2月更文挑战第19天】Kafka【环境搭建 01】kafka_2.12-2.6.0 单机版安装+参数配置及说明+添加到service服务+开机启动配置+验证+chkconfig配置说明(一篇入门kafka)
42 1
|
2天前
|
消息中间件 存储 运维
Kafka重要配置参数全面解读(重要)
Kafka重要配置参数全面解读(重要)
20 0
|
24天前
|
监控 网络协议 Linux
Linux 命令大全 & CentOS常用运维命令
Linux 命令大全 & CentOS常用运维命令
134 0
|
24天前
|
关系型数据库 MySQL 数据库
虚拟机Linux-Centos系统网络配置常用命令+Docker 的常用命令
虚拟机Linux-Centos系统网络配置常用命令+Docker 的常用命令
28 0

热门文章

最新文章