如果是为了简单的调试，安装单机版kafka也是可以的。不过生产环境肯定是需要搭建集群，所以这篇博客是记录在本地虚拟机尝试搭建kafka的集群，为生产做准备。
环境介绍：

一、环境介绍

1. 三台VMware上的centos7：
   192.168.220.129
   192.168.220.130
   192.168.220.131
   先在第一台上做好前期准备工作，然后克隆生成后两台
2. jdk8：java -version
   我用的是系统自带的：

openjdk version "1.8.0_191"
OpenJDK Runtime Environment (build 1.8.0_191-b12)
OpenJDK 64-Bit Server VM (build 25.191-b12, mixed mode)

二、Zookeeper集群

kafka依赖zookeeper，所以我们需要先搭建zookeeper集群。

1. 创建文件夹

mkdir zookeeper
mkdir zookeeper/zkdata
mkdir zookeeper/zkdatalog

1. 下载zookeeper
   进入文件夹：cd zookeeper
   推荐直接在阿里巴巴开源镜像站下载
   wget https://mirrors.aliyun.com/apache/zookeeper/zookeeper-3.4.13/zookeeper-3.4.13.tar.gz
2. 解压
   tar -zxvf zookeeper-3.4.13.tar.gz
3. 复制一份配置文件：zoo.cfg

cd zookeeper-3.4.13/conf
cp zoo_sample.cfg zoo.cfg

1. 编辑配置文件vim zoo.cfg

#心跳间隔
tickTime=2000
#其他服务器连接到Leader时，最长能忍受的心跳间隔数：10*2000 = 20秒
initLimit=10
#发送消息时，多长忍受心跳间隔数：5*2000 = 10秒
syncLimit=5
#快照日志
dataDir=/root/myapp/zookeeper/zkdata
#事务日志
dataLogDir=/root/myapp/zookeeper/zkdatalog
#客户端连接zookeeper服务器的端口
clientPort=2181
#可以待后续克隆完剩下两台后，再写上其他两台的ip
server.1=192.168.220.129:2888:3888
server.2=192.168.220.130:2888:3888
server.3=192.168.220.131:2888:3888

1. 克隆该虚拟机，你也可以把kafka的准备工作做好后，再统一克隆
   如果你已经有多台机子，则可跳过该步骤
   软件VMware中，关闭虚拟机后，右键点击对应虚拟机>>管理>>克隆
   为了不必要的麻烦，记得选择：
   克隆虚拟机
2. 修改每台虚拟机里的zookeeper的zoo.cfg文件中的server ip信息
   查看本机ip：ifconfig
3. 创建myid文件，对应zoo.cfg里的server.1、server.2、server.3

#server1
echo "1" > xxxxxxxx/zookeeper/zkdata/myid
#server2
echo "2" > xxxxxxxx/zookeeper/zkdata/myid
#server3
echo "3" > xxxxxxxx/zookeeper/zkdata/myid

1. 启动

• 每一台：./bin/zkServer.sh start
• 查看状态：./bin/zkServer.sh status
  查看状态
• 如果状态异常，可以查看当前目录下的zookeeper.out日志文件

1. 停止：./bin/zkServer.sh stop

三、kafka集群

1. 创建文件夹mkdir kafka
2. 下载
   与zookeeper一样，推荐到阿里巴巴开源镜像站下载
   wget https://mirrors.aliyun.com/apache/kafka/2.1.0/kafka_2.12-2.1.0.tgz
3. 解压
   tar -zxvf kafka_2.12-2.1.0.tgz
4. 修改配置文件
   vim kafka_2.12-2.1.0/config/server.properties
   主要修改如下：

#分别是1/2/3！！！！！！！！！！！
broker.id=3
#对应ip
listeners=PLAINTEXT://192.168.220.131:9092
#对应ip
advertised.listeners=PLAINTEXT://192.168.220.131:9092
#zookeeper集群地址
zookeeper.connect=192.168.220.129:2181,192.168.220.130:2181,192.168.220.131:2181

1. 启动
   ./bin/kafka-server-start.sh -daemon ./config/server.properties

• 控制台查看日志
  ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.220.129:9092 --topic kafka-topic
• 如果正常的话，则应该不会有任何输出信息
  如果不正常，可查看详细日志文件：
  tail -fn 100 logs/server.log

1. 停止
   ./bin/kafka-server-stop.sh
   OK！

四、Kafka Topic分区

• 在后续实践中才了解到分区的事情，简而言之，为了让不同的消息往不同的kafka server发送消息，实现负载均衡
• 我们在创建Topic时，需要指定分区数
  ./bin/kafka-topics.sh --create --zookeeper zookeeper_ip1:2181,zookeeper_ip2:2181,zookeeper_ip3:2181 --replication-factor 3 --partitions 3 --topic your_topic
  建议分区数为server个数的整数倍
• 或者修改分区
  ./bin/kafka-topics.sh --alter --zookeeper 10.12.4.38:2181,10.12.4.39:2181,10.12.4.40:2181 --topic kafka-new-topic --partitions 3
• client在发送消息时，指定往某个分区发送
• 查看指定分区的日志，例如分区：2
  ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.220.129:9092 --topic your_topic --partition 2

五、其它常用命令

• 查看所有topic
  bin/kafka-topics.sh --zookeeper node01:2181 --list
• 查看某个topic信息
  bin/kafka-topics.sh --zookeeper node01:2181 --describe --topic your_topic
• 控制台发送消息
  bin/kafka-console-producer.sh --broker-list {kafka-ip1}:9092 --topic your_topic