kafka集群搭建测试

一、环境

zookeeper集群，请看前面我的zookeeper搭建文档
master:192.168.31.131
slave1:192.168.31.132
slave2:192.168.31.133
Scala 2.11

二、安装

1.下载
下载官网：http://kafka.apache.org/downloads
下载版本：与自己安装的Scala版本对应的版本，个人习惯是下载最新版本的前一版
kafka_2.11-0.10.0.1.tgz
2.安装

tar -xzf  kafka_2.11-0.10.0.1.tgz
cp kafka_2.11-0.10.0.1.tgz /home/cms/kafka

3.配置环境变量
即path、classpath,意义不大，可不配置
4.修改配置文件kafka/config/server.properties

###必须修改的值
broker.id=0 #整个集群内唯一id号，整数，一般从0开始
listeners=PLAINTEXT://192.168.31.131:9092 #协议、当前broker机器ip、端口
port=9092 #broker端口
host.name=192.168.31.131 #broker 机器ip
log.dirs=/home/cms/kafka/kafkalogs #kafka存储数据的目录
zookeeper.connect=192.168.31.131:2181,192.168.31.132:2181,192.168.31.133:2181 #zookeeper 集群列表
###下面根据实际情况而定
# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=502400 #将buffer改大一点
# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=502400
###更多配置可参考官网

5.在kafka的目录下，建立kafka存储数据的目录
mkdir kafkalogs

三、其他节点安装配置

1.将kafka文件复制到其他节点

scp -r kafka cms@slave2:/home/cms```
**2.修改其他节点下配置文件kafka/config/server.properties **
主要改以下几项
slave1：

broker.id=1 #整个集群内唯一id号，整数，一般从0开始
listeners=PLAINTEXT://192.168.31.132:9092 #协议、当前broker机器ip、端口
port=9092 #broker端口
host.name=192.168.31.132 #broker 机器ip

slave2：

broker.id=1 #整个集群内唯一id号，整数，一般从0开始
listeners=PLAINTEXT://192.168.31.133:9092 #协议、当前broker机器ip、端口
port=9092 #broker端口
host.name=192.168.31.133 #broker 机器ip

####四、集群启动
**1.启动zookerper**
在每个节点下，都启动

zkServer.sh start

**2.启动kafka进程，在每个节点的kafka/bin目录下**

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

jps #查看进程的启动情况

![](http:https://img.haomeiwen.com/i1908836/060729d68b34b6a3.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

![](https://img.haomeiwen.com/i1908836/a487fdb597b48cf1.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
>"-daemon" 参数代表以守护进程的方式启动kafka server。

####五、集群测试
**1.创建主题**

kafka-topics.sh --create --zookeeper 192.168.31.131:2181,192.168.31.132:2181,192.168.31.133:2181 --replication-factor 3 --partitions 3 --topic test5```

--zookeeper : zookeeper集群列表，用英文逗号分隔。可以不用指定zookeeper整个集群内的节点列表，只指定某个或某几个zookeeper节点列表也是你可以的
replication-factor ： 复制数目，提供failover机制；1代表只在一个broker上有数据记录，一般值都大于1，代表一份数据会自动同步到其他的多个broker，防止某个broker宕机后数据丢失。
partitions ： 一个topic可以被切分成多个partitions，一个消费者可以消费多个partitions，但一个partitions只能被一个消费者消费，所以增加partitions可以增加消费者的吞吐量。kafka只保证一个partitions内的消息是有序的，多个一个partitions之间的数据是无序的。

2.启动生产者

kafka-console-producer.sh --broker-list 192.168.31.131:9092 --topic test5

--broker-list : 值可以为broker集群中的一个或多个节点

3.启动消费者

kafka-console-consumer.sh  --zookeeper 192.168.31.131:2181,192.168.31.132:2181,192.168.31.133:2181 --topic test5 --from-beginning

--zookeeper : 值可以为zookeeper集群中的一个或多个节点
--from-beginning 表示从开始第一个消息开始接收

我们可以发现，在生产者下输入一个消息，回车后可以在消费者下输出，说明我们集群搭建的没问题
生产者：

消费者：

关闭进程：
ps -ef | grep kafka

kill 2853

六、其他命令

1.查看topic

kafka-topics.sh --list --zookeeper 192.168.31.131:2181,192.168.31.132:2181,192.168.31.133:2181

2.查看topic详情

 kafka-topics.sh --describe --zookeeper 192.168.31.131:2181,192.168.31.132:2181,192.168.31.133:2181 --topic test5

状态说明：test有三个分区分别为1、2、3，分区0的leader是3（broker.id），分区0有三个副本，并且状态都为lsr（ln-sync，表示可以参加选举成为leader）。

3.删除topic
在config/server.properties中加入delete.topic.enable=true并重启服务，在执行如下命令

kafka-topics.sh --delete --zookeeper 192.168.31.131:2181,192.168.31.132:2181,192.168.31.133:2181 --topic test5

七、参考文档

kafka官网
Kafka集群搭建
kafka环境搭建2-broker集群+zookeeper集群