【Linux】ubuntu下kafka集群环境搭建，kafka命令

本是古典何须时尚 2022-07-17 00:16 240阅读 0赞

http://blog.csdn.net/catshitone/article/details/50462023

１.下载kafka正式版，本文使用kafka\_2.10-0.8.2.2（在此特别说明下，kafka\_2.9.2-0.8.1.1.tgz由于版本相对较低的问题，好多.sh文件时缺少的，比如kafka-topics.sh，所以建议大家在安装的时候不要使用太老的版本，问题较多，并且启动kafka-server-start的时候还报错，始终无法启动成功～）

官网下载链接如下：[http://kafka.apache.org/downloads.html][http_kafka.apache.org_downloads.html]

２．解压tar包

命令如下：tar -zxvf kafka\_2.10-0.8.2.2.tar.gz

3.修改server.properties相关配置

进入kafka根目录下的config/server.properties，主要修改的３个参数如下：

broker.id=0 //broker的标识（正数），集群内各个broker.id不能重复。

port=9092 //端口号，单节点内各个port不能重复(为了方便看最好不同节点端口也不要重复)。

log.dir =/tmp //存储log的地方（数据文件）。可根据自己的日志目录进行修改

以上配置时在单机上的配置，多台机器的配置要根据ｉｐ进行相关配置；

４．修改config下面的log4j.properties、zookeeper.properties的日志目录；

log4j.properties下面的kafka.logs.dir=logs，可根据自己需要修改成自己的目录；

zookeeper.properties下面的dataDir=/tmp/zookeeper，可根据需要修改成自己的目录

５．验证是否安装成功

**（１） start server**

启动一个单节点的ｚｏｏｋｅｅｐｅｒ，命令如下：>bin/zookeeper-server-start.sh config/zookeeper.properties

**　准备启动一个３个ｂｒｏｋｅｒ节点的ｋａｆｋａ集群，因此做如下配置：**

$ cp config/server.properties config/server-1.properties  
　　　$ cp config/server.properties config/server-2.properties

并做如下修改：

config/server-1.properties:  
　　　　 broker.id=1  
　　　　port=9093  
　　　　 log.dir=/tmp/kafka-logs-1

config/server-2.properties:  
　　　　 broker.id=2  
　　　　 port=9094  
　　　　 log.dir=/tmp/kafka-logs-2

说明：  
　　　broker.id: broker节点的唯一标识  
　　　port: broker节点使用端口号  
　　　log.dir: 消息目录位置

**启动3个broker节点**  
　　　$ JMX\_PORT=9997 bin/kafka-server-start.sh config/server-1.properties &  
　　　$ JMX\_PORT=9998 bin/kafka-server-start.sh config/server-2.properties &  
　　　$ JMX\_PORT=9999 bin/kafka-server-start.sh config/server.properties &

标红的内容可根据自己的需要进行配置；

启动单个ｂｒｏｋｅｒ节点的ｋａｆｋａ集群，命令如下：>bin/kafka-server-start.sh config/server.propertie

第一行时启动ｋａｆｋａ自带的ｚｏｏｋｅｅｐｅｒ

**（２）create a topic**

$bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

$bin/kafka-topics.sh --list --zookeeper localhost:2181

topic: 3test partition: 0 leader: 2 replicas: 2,1,0 isr: 2,1,0  
　topic: test partition: 0 leader: 0 replicas: 0 isr: 0  
　topic: test\_topic partition: 0 leader: 1 replicas: 0,1,2 isr: 1,2,0  
　说明：  
　partiton： partion id，由于此处只有一个partition，因此partition id 为0  
　leader：当前负责读写的lead broker id  
　relicas：当前partition的所有replication broker list  
　isr：relicas的子集，只包含出于活动状态的broker

（３）send messag e

$bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

**（４）start a consumer**

$bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

kafka-console-producer.sh和kafka-console-cousumer.sh只是系统提供的命令行工具。这里启动是为了测试是否能正常生产消费，验证流程正确性，在实际开发中还是要自行开发自己的生产者与消费者。

**（５）start a produce**

$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

这样就可以的命令行输入信息，可以看到ｃｏｎｓｕｍｅｒ可以获得同样的信息～

**6.干掉follow broker**

杀掉一个非leader broker命令如下:server-1.properties里面的ｂｒｏｋｅｒ　＝１

$ pkill -9 -f server-1.properties

查看topic：  
　　　$ bin/kafka-list-topic.sh --zookeeper localhost:2181  
　　　topic: 3test partition: 0 leader: 2 replicas: 2,1,0 isr: 2,0  
　　　topic: test partition: 0 leader: 2 replicas: 0,1,2 isr: 2,0  
　　　此时，存活的broker只有2,0  
　　　测试：produce发送消息，consumer能正常接收到

**7.继续干掉leader broker**

干掉leader broker后，连续查看topic状态  
$ pkill -9 -f server-2.properties  
$ bin/kafka-list-topic.sh --zookeeper localhost:2181  
topic: test partition: 0 leader: 2 replicas: 2,1,0 isr: 2,0  
  
$ bin/kafka-list-topic.sh --zookeeper localhost:2181  
topic: test partition: 0 leader: 2 replicas: 2,1,0 isr: 2,0  
  
$ bin/kafka-list-topic.sh --zookeeper localhost:2181  
topic: test partition: 0 leader: 0 replicas: 2,1,0 isr: 0  
  
$ bin/kafka-list-topic.sh --zookeeper localhost:2181  
 topic: test partition: 0 leader: 0 replicas: 2,1,0 isr: 0  
  
杀掉leader broker过了一会，broker 0成为新的leader broker  
测试：produce发送消息，consumer能正常接收到

**8．kafka常用命令**

创建topic

/kafka-topics.sh --create --zookeeper 192.168.153.128:2181 --replication-factor 1--partitions 1 --topic test123

查看topic信息

./kafka-topics.sh --describe --topic test123 --zookeeper 192.168.153.128:2181

修改topic分区

./kafka-topics.sh --alter --topic test123 --partitions 2 --zookeeper 192.168.153.128:2181

删除topic

./kafka-run-class.shkafka.admin.DeleteTopicCommand --topic test1 --zookeeper192.168.35.122:2181,192.168.35.123:2181

只是删除了zookeeper中元数据，数据文件还需手动删除。

[http_kafka.apache.org_downloads.html]: http://kafka.apache.org/downloads.html