提前准备：
安装jdk-8u201-linux-x64.tar.gz环境
安装zk集群，也可以采用kafka自带的zk，这里我使用的是自己创建的zk集群
服务器IP:
zk1:192.168.1.2
zk2:192.168.1.3
zk3:192.168.1.4
kafka:192.168.1.5
下载并解压：
wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz
tar -xf kafka_2.12-2.3.0.tgz
如下加入到.bash_profile
export KAFKA_HOME=/opt/app/kafka_2.12-2.3.0
export PATH=$PATH:$KAFKA_HOME/bin
安装：
cd kafka_2.12-2.3.0
修改/opt/app/kafka_2.12-2.3.0/config/server.properties配置
broker.id=0
listeners=PLAINTEXT://192.168.1.5:9092
log.dirs=/tmp/kafka-logs
zookeeper.connect=192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181
cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties
编辑这些新文件并设置一下属性
config/server-1.properties
broker.id=1
listeners=PLAINTEXT://192.168.1.5:9093
log.dirs=/tmp/kafka-logs-1
zookeeper.connect=192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181
config/server-1.properties
broker.id=2
listeners=PLAINTEXT://192.168.1.5:9094
log.dirs=/tmp/kafka-logs-2
zookeeper.connect=192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181
该broker.id属性是群集中每个节点的唯一且永久的名称。我们必须覆盖端口和日志目录，因为我们在同一台机器上运行这些，并且我们希望让所有代理尝试在同一端口上注册或覆盖彼此的数据。
启动
nohup bin/kafka-server-start.sh config/server.properties &
nohup bin/kafka-server-start.sh config/server1.properties &
nohup bin/kafka-server-start.sh config/server2.properties &
现在我们创建一个复制因为为3的新主题：
bin/kafka-topics.sh --create --bootstrap-server 192.168.1.5:9092 --replication-factor 3 --partitions 1 --topic my-replicated-topic
现在我们有一个集群，我们怎么知道哪个经纪人正在做什么？要查看运行“describe topics”命令：
bin/kafka-topics.sh --describe --bootstrap-server 192.168.1.5:9092 --topic my-replicated-topic
“leader”是负责给定分区的所有读写的节点。每个节点将成为随机选择的分区部分的领导者。
“replicas”是复制此分区日志的节点列表，无论它们是否为领导者，或者即使它们当前处于活动状态。
“isr”是“同步”复制品的集合。这是副本列表的子集，该列表当前处于活跃状态并且已经被领导者捕获。
让我们向新主题发布一些消息：
bin/kafka-console-producer.sh --broker-list 192.168.1.5:9092 --topic my-replicated-topic
>a
>b
现在让我们来使用这些消息：
bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.5:9092 --from-beginning --topic my-replicated-topic
a
b