一.前期准备
1.1 Win7官网下载kafka包
本文使用版本kafka_2.10-0.10.1.0.tgz
1.2 配置jdk、scala、zookeeper
jdk1.7:linux jdk安装和配置
scala2.10.6:linux scala安装和配置
zookeeper3.4.9:zookeeper3.49集群安装和配置
jdk,scala,kafka版本要对应
1.3 centos7集群服务器
主机名 系统 IP地址
master centos7 192.168.32.128
slave01 centos7 192.168.32.131
slave02 centos7 192.168.32.132
二.kafka集群搭建
以下操作只针对master主机服务器，其他主机服务器类似。
2.1 上传kafka包至 /opt/software目录
2.2 解压和拷贝kafka至 /usr/local/kafka
cd /opt/software
tar -zxvf kafka_2.10-0.10.1.0.tgz
cp -r kafka_2.10-0.10.1.0 /usr/local/kafka
三.kafka集群配置
3.1 server.properties文件配置
进入kafka中的config目录
vi server.properties
#switch to enable topic deletion or not, default value is false
delete.topic.enable=true
#本机对应的ip地址
listeners=PLAINTEXT://192.168.32.128:9092
log.dirs=/tmp/kafka-logs
zookeeper.connect=192.168.32.128:2181,192.168.32.132:2181,192.168.32.131:2181
注意：相关配置项不要重复。
3.2 新增/tmp/kafka-logs
mkdir -p /tmp/kafka-logs
至此 master主机服务器kafka已搭建完成。
3.3 其他服务器kafka搭建
搭建192.168.32.131/slave01和192.168.32.132/slave02服务器的kafka
拷贝master服务器的kafka至slave01和slave02的 /usr/local/目录
master服务器：
cd /usr/local
scp -r kafka root@192.168.32.131:/usr/local/
slave01服务器：
类似3.1配置server.properties
注意：
broker.id=1,要与master中broker.id区别
listeners=PLAINTEXT://192.168.32.131:9092,需要改成本机对应ip
slave02服务器配置类似。
四.kafka集群测试
4.1 测试命令
Step 1: Start the server
后台方式启动，推荐第一次配置的新手不要加入-daemon参数，看看控制台输出的是否有success.
bin/kafka-server-start.sh -daemon config/server.properties
Step 2: Create a topic(replication-factor一定要大于1，否则kafka只有一份数据，leader一旦崩溃程序就没有输入源了，分区数目视输入源而定)
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic topicTest
Step 3: Describe a topic
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic topicTest
step 4: list the topic
bin/kafka-topics.sh --list --zookeeper localhost:2181
step 5: send some message
bin/kafka-console-producer.sh --broker-list localhost:2181 --topic topicTest
step 6: start a consumer
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic topicTest --from-beginning
step 7: delete a topic
要事先在 serve.properties 配置 delete.topic.enable=true
bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topicTest
# 如果仍然只是仅仅被标记了删除(zk中并没有被删除)，那么启动zkCli.sh,输入如下指令
rmr /brokers/topics/topicTest
4.2 集群测试
step1:启动zookeeper集群
step2:启动kafka集群
三台服务器分别运行启动命令
bin/kafka-server-start.sh -daemon config/server.properties
jps查看进程
注意：kafka是kafka进程,QuorumPeerMain是zookeeper进程
step3. 创建主题和查看主题
bin/kafka-topics.sh --create --zookeeper 192.168.32.128:2181,192.168.32.131:2181,192.168.32.132:2181 --replication-factor 3 --partitions 3 --topic topicTest
bin/kafka-topics.sh --list --zookeeper 192.168.32.128:2181,192.168.32.131:2181,192.168.32.132:2181
step4. 启动消息生产者和消息消费者
master服务器
bin/kafka-console-producer.sh --broker-list 192.168.32.128:9092,192.168.32.131:9092,192.168.32.132:9092 --topic topicTest
slave01或slave02服务器
bin/kafka-console-consumer.sh --zookeeper 192.168.32.128:2181,192.168.32.131:2181,192.168.32.132:2181 --topic topicTest --from-beginning
master服务器输入信息，slave01或slave02会显示master输的的信息
