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

2021-08-17 10:55:06 浏览数 (1)


安装JDK

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

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

代码语言:javascript复制
[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版本的都可以。

代码语言:javascript复制
[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里。

代码语言:javascript复制
[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客户端退出后,就会停止服务

代码语言:javascript复制
[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下的节点信息

代码语言:javascript复制
[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 ,回车可以看到参数说明。

代码语言:javascript复制
[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]# 

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

代码语言:javascript复制
--create                                 Create a new topic.                   

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

代码语言:javascript复制
[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

代码语言:javascript复制
[root@artisan bin]# ./kafka-topics.sh --list --zookeeper 192.168.18.130:2181
artisan
[root@artisan bin]# 

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

删除主题

代码语言:javascript复制
# 删除
[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集群中。

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

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

代码语言:javascript复制
[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同样也携带了一个命令行客户端,会将获取到内容在命令中进行输出,默认是消费最新的消息.

代码语言:javascript复制
./kafka-console-consumer.sh --bootstrap-server 192.168.18.130:9092   --topic artisan

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

代码语言:javascript复制
[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的终端窗口上显示出来。

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


查看组名

代码语言:javascript复制
[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]# 

查看消费者的消费偏移量

代码语言:javascript复制
[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

代码语言:javascript复制
[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]# 

开启两个生产者

代码语言:javascript复制
[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
>

消费多主题 如下

代码语言:javascript复制
[root@artisan bin]# ./kafka-console-consumer.sh --bootstrap-server  192.168.18.130:9092 --whitelist "artisan|xiaogongjiang"
send from xiaogongjiang
send from artisan

单播消费

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

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

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

生产者

代码语言:javascript复制
[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

代码语言:javascript复制
[root@artisan bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.18.130:9092 --consumer-property group.id=artisanGroup --topic artisan

消费者2

代码语言:javascript复制
[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 消费组, 结果两个客户端都能收到消息.

生产者

代码语言:javascript复制
[root@artisan bin]# ./kafka-console-producer.sh --broker-list 192.168.18.130:9092 --topic artisan
>messge artisan jajaja
>

消费者1 属于 anotherArtisanGroup消费组

代码语言:javascript复制
[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 消费组

代码语言:javascript复制
[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集群吧。

0 人点赞