Kafka版本：kafka_2.12-2.1.1.tgz jdk版本：JDK1.8以上
版本号解释：Scala版本号：2.12 Kafka版本号：2.1.1
Kafka集群安装步骤：
1.下载安装包
wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.1.1/kafka_2.12-2.1.1.tgz
2.解压缩
tar -zxvf kafka_2.12-2.1.1 -C /usr/local (-C 表示解压缩到指定目录下)
mv kafka_2.11-2.1.0 kafka(更名)
3.修改配置文件
cd /usr/local/kafka/config 进入kafka中config文件夹
修改server.properties文件
备注：必须修改部分
1.broker的全局唯一编号 broker.id=0
2.kafka运行日志存放的路径(不是必须) log.dirs=随意设置
3.broker使用zookeeper信息 zookeeper.connect=192.168.***.***:2181,192.168***.***:2181,192.168.***.***:2181
4.此处的listeners为本机IP advertised.listeners=PLAINTEXT://192.168.***.***:9092
附：server.properties文件属性介绍
#broker的全局唯一编号，不能重复
broker.id=0
#用来监听链接的端口，producer或consumer将在此端口建立连接
port=9092
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的线程数量
num.io.threads=89
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接受套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志存放的路径
log.dirs=/export/servers/logs/kafka
#topic在当前broker上的分片个数
num.partitions=2
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#滚动生成新的segment文件的最大时间
log.roll.hours=168
#日志文件中每个segment的大小，默认为1G
log.segment.bytes=1073741824
#周期性检查文件大小的时间
log.retention.check.interval.ms=300000
#日志清理是否打开
log.cleaner.enable=true
#broker需要使用zookeeper保存meta数据
zookeeper.connect=192.168.***.***:2181,192.168.***.***:2181,192.168.***.***:2181
#zookeeper链接超时时间
zookeeper.connection.timeout.ms=6000
#partion buffer中，消息的条数达到阈值，将触发flush到磁盘
log.flush.interval.messages=10000
#消息buffer的时间，达到阈值，将触发flush到磁盘
log.flush.interval.ms=3000
#删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除
delete.topic.enable=true
#此处的listener为本机IP(重要)，如果不改，则客户端会抛出：Producer connection to localhost:9092 unsuccessful错误！
advertised.listeners=PLAINTEXT://192.168.***.***:9092
#配置此进行删除时，会将topic删除(不配置该项，在删除topic时，置灰对其进行标记为删除，并没有实际删除)
delete.topic.enable=true4.zookeeper(集群)的安装
Kafka集群，需要zookeeper的支持，通过zookeeper来完成Kafka集群的高可用。ZooKeeper集群的安装，请移步：ZooKeeper集群的安装
如果没有搭建 zookeeper 环境，可以直接运行kafka 内嵌的 zookeeper。
命令：bin/zookeeper-server-start.sh config/zookeeper.properties &
5.分发安装包到其他服务器
scp -r /usr/localkafka root@192.168.***.***:/usr/local/
scp -r /usr/local/kafka root@192.168.***.***:/usr/local/（202,203为kafka集群其他两台节点）
6.再次修改其他机器的配置文件(重要)
①依次修改服务器上的配置文件的broker.id，分别是0，1，2，不得重复
②依次修改advertised.listeners中的IP地址（集群修改）
7.启动集群
依次在各节点启动Kafka
【根据配置文件，前台启动】./kafka-server-start.sh -daemon /usr/local/kafka_2.11-2.1.1/config/server.properties
【根据配置文件，后台启动】./kafka-server-start.sh -daemon /usr/local/kafka_2.11-2.1.1/config/server.properties &
【根据配置文件，后台启动 + 日志显示】./kafka-server-start.sh /usr/local/kafka_2.11-2.1.1/config/server.properties &
备注：
①kafka-server-start.sh为启动命令，位于bin目录下
②/usr/local/kafka/config/server.properties为kafka的配置文件路径
8.至此，集群启动完毕
9.此时，我们通过命令 ./zkCli.sh 连接 ZooKeeper，来查看相关节点信息
①我们能够看到ZooKeeper多了一个brokers节点；
②在brokers节点下，多了一个ids节点，我们可以看到三个配置的brokers.id已经都成功连接至ZooKeeper
10.至此，才能说明Kafka集群真正安装完成
提示：如遇到如下错误：请开放9092端口，或者关闭防火墙，线上环境建议开放端口(开放端口，请参考：Linux开放指定端口)
***.***:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
切记，也需要开放ZooKeeper的2888,3888端口
创建kafka的topic
./kafka-topics.sh --create --topic dic --partitions 4 --replication-factor 2 --zookeeper 192.168.***.***:2181,192.168.***.***:2181,192.168.***.***:2181
写数据到kafka （启动kafka topic）
./kafka-console-producer.sh --broker-list 192.168.***.***:9092,192.168.***.***:9092,192.168.***.***:9092 --topic dic
读取kafka topic的数据
./kafka-console-consumer.sh --bootstrap-server 192.168.***.***:9092,192.168.***.***:9092,192.168.***.***:9092 --from-beginning --topic dic
删除topic
./kafka-topics.sh --delete --zookeeper 192.168.***.***:2181,192.168.***.***:2181,192.168.***.***:2181 --topic dic