一、环境
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 #查看进程的启动情况
>"-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
