一、Kafka集群

二、搭建两台服务器

笔者的两台服务器IP：

服务器	IP
1号	192.168.11.59
2号	192.168.11.235

2.1、 zookeeper部署

zookeeper先只部署一台，在1号机器（192.168.11.59）上启动zookeeper。

sh zookeeper-server-start.sh -daemon ../config/zookeeper.properties

注意：shell脚本在kafka安装目录的bin文件夹中。

2.2、 启动1号机器的broker

（1）修改server.properties（在config目录）。

broker.id=0

zookeeper.connect=192.168.11.59:2181

修改broker.id（也可以改为-1，自动分配）和配置对应的zookeeper ip地址。

（2）启动kafka。

sh kafka-server-start.sh -daemon ../config/server.properties

默认端口为：9092，可以通过命令lsof -i:9092查看kafka是否启动成功。

$ lsof -i:9092
COMMAND   PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java    84858  fly  152u  IPv6 802800      0t0  TCP *:9092 (LISTEN)
java    84858  fly  168u  IPv6 813310      0t0  TCP localhost:50918->fly-virtual-machine:9092 (CLOSE_WAIT)
java    84858  fly  172u  IPv6 810764      0t0  TCP localhost:50920->fly-virtual-machine:9092 (CLOSE_WAIT)
java    84858  fly  180u  IPv6 812452      0t0  TCP localhost:50924->fly-virtual-machine:9092 (ESTABLISHED)
java    84858  fly  181u  IPv6 810766      0t0  TCP fly-virtual-machine:9092->localhost:50924 (ESTABLISHED)
java    84858  fly  186u  IPv6 815339      0t0  TCP 192.168.11.59:9092->192.168.11.59:42758 (ESTABLISHED)
java    84858  fly  187u  IPv6 815340      0t0  TCP fly-virtual-machine:9092->localhost:50978 (ESTABLISHED)
java    84858  fly  188u  IPv6 815341      0t0  TCP fly-virtual-machine:9092->localhost:50980 (ESTABLISHED)
java    88485  fly   89u  IPv6 813836      0t0  TCP 192.168.11.59:42758->192.168.11.59:9092 (ESTABLISHED)
java    88485  fly   90u  IPv6 813837      0t0  TCP localhost:50978->fly-virtual-machine:9092 (ESTABLISHED)
java    88485  fly   91u  IPv6 813838      0t0  TCP localhost:50980->fly-virtual-machine:9092 (ESTABLISHED)

2.3、 启动2号机器的broker

（1）修改server.properties（在config目录）。

broker.id=1

zookeeper.connect=192.168.11.59:2181

修改broker.id（也可以改为-1，自动分配）和配置对应的zookeeper ip地址。
（2）启动kafka。

sh kafka-server-start.sh -daemon ../config/server.properties

默认端口为：9092，可以通过命令lsof -i:9092查看kafka是否启动成功。

2.4、查看kafka集群

（1）创建主题。

sh kafka-topics.sh --create --zookeeper 192.168.11.59:2181 -replication-factor 2 --partitions 2 --topic kafka-2

（2）查看主题。

sh kafka-topics.sh --describe --zookeeper 192.168.11.59:2181 --topic kafka-2

显示消息：

Topic:kafka-2	PartitionCount:2	ReplicationFactor:2	Configs:
	Topic: kafka-2	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: kafka-2	Partition: 1	Leader: 0	Replicas: 0,1	Isr: 0

2.5、测试集群

开启一个生产者，两个消费者。当两个消费者同属一个消费组开启后，消费者轮流收到发送者的数据。

（1）生产者：

sh kafka-console-producer.sh --broker-list 192.168.11.59:9092 --topic kafka-2

（2）消费者：

sh kafka-console-consumer.sh --bootstrap-server 192.168.11.59:9092 --topic kafka-2

默认是group 0，也可以指定，比如：

sh kafka-console-consumer.sh --bootstrap-server 192.168.31.249:9092 --topic kafka-2 --group 0 --from-beginning

kafka-console-consumer.sh部分支持的参数：

参数	值类型	说明	有效值
–topic	string	被消费的topic
-partition	integer	指定分区 除非指定’–offset’，否则从分区结束(latest)开始消费
–offset	string	执行消费的起始offset位置 默认值:latest	latest、earliest
–consumer-property	string	将用户定义的属性以key=value的形式传递给使用者
–consumer.config	string	消费者配置属性文件 请注意，[consumerproperty]优先于此配置
–from-beginning		从存在的最早消息开始，而不是从最新消息开始
–group	string	指定消费者所属组的ID

总结

本节演示了一个zookeeper加两个broker的部署，实际使用总只是是三个zookeeper，三个broker。部署方式相同。

后言

本专栏知识点是通过<零声教育>的系统学习，进行梳理总结写下文章，对c/c++linux系统提升感兴趣的读者，可以点击链接,详细查看详细的服务:C/C++服务器课程 。