最近公司项目需要用到kafka消息队列，所以特地花了一个周末的时间对kafka的搭建以及使用做了一个了解，特地记录一下。
kafka搭建需要依赖于zookeeper作为注册中心，前面两篇文已经介绍了zookeeper的环境搭建，现在将记录一下kafka的搭建，同样也是选取三个节点
如下：
Node1:192.168.153.133
Node2:192.168.153.137
Node3:192.168.153.138
开始安装kafka，参照官网地址：http://kafka.apachecn.org/quickstart.html
执行命令：
/#进入下载目录 cd /usr/opt/ #下载kafka wget http://archive.apache.org/dist/kafka/1.0.0/kafka_2.11-1.0.0.tgz #解压文件并且重命名 tar -zxvf kafka_2.11-1.0.0.tgz && mv kafka_2.11-1.0.0 kafka
编辑配置文件（三台机器都是如此配置，记得修改ip以及broker.id）：
#修改日志存储地址 log.dirs=/usr/opt/kafka/data #节点id，用于区分子节点，就算是节点ip改变了，也能找到此节点 broker.id=1 #将使用外部的zookeeper配置 zookeeper.connect=192.168.153.133:2008,192.168.153.137:2008,192.168.153.138:2008 #指定主机地址，区分localhost，避免出错 host.name=192.168.153.137 #监听地址 listeners = PLAINTEXT://192.168.153.137:9092 #添加此项是为了api能够监听到此服务 advertised.listeners = PLAINTEXT://192.168.153.137:9092
配置kafka环境变量：
#编辑环境配置
vim /etc/profile
加入以下配置：
export PATH=$PATH:$ZOOKEEPER_HOME/bin
export KAFKA_HOME=/usr/opt/kafka
然后生效配置文件：
#环境配置文件生效
source /etc/profile
后台启动kafka
./bin/kafka-server-start.sh -daemon config/server.properties
检查kafka是否启动成功：
jps
启动成功之后显示：
更多命令查看官网介绍：http://kafka.apachecn.org/quickstart.html
创建生产者
bin/kafka-topics.sh --create --zookeeper 192.168.153.137:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
测试：
查看主题列表 > bin/kafka-topics.sh --list --bootstrap-server 192.168.153.137:9092 启动控制台消费者 > bin/kafka-console-consumer.sh --bootstrap-server 192.168.153.137:9092 --topic my-replicated-topic--from-beginning 启动控制台消费者 > bin/kafka-console-consumer.sh --bootstrap-server 192.168.153.138:9092 --topic my-replicated-topic--from-beginning
在生产者发送一条消息，可以再消费者都接收到，成功！
另外出现Connection to node 1 (/192.168.153.133:9092) could not be established. Broker may not be available.错误的时候，可能是防火墙未关闭，或者是端口访问不同，注意检查，
还有就是kafka 发送消息 报错 topic test with key: null
可能是9092端口未打开，此时只要打开端口并关闭防火墙就行
firewall-cmd --zone=public --add-port=9092/tcp --permanent service firewalld stop
然后重启kafka即可