一.前期准备

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输的的信息