因为之前公司业务都是使用mqtt接收数据，随着设备的增加，公司觉得用kafka集群来实现会更好 下面是我写一个demo 仅供参考

一、安装jdk

没有安装的可以百度去看看怎么安装的 不行的话在私聊我！！！！

二、搭建zookeeper集群

因为我只有一台Linux服务器，所以在一台机器上创建三个zk实例，通过设置不同的端口号，搭建一个zookeeper伪集群。

1、下载zookeeper安装包

地址：https://archive.apache.org/dist/zookeeper/，我这里下的是apache-zookeeper-3.6.3-bin.tar.gz版本。

2、新建一个zookeeper-cluster目录，将安装包上传到zookeeper-cluster目录下

  cd /usr/local/
  mkdir zookeeper-cluster

3、解压安装包

cd zookeeper-cluster/
tar -zxvf  apache-zookeeper-3.6.3-bin.tar.gz

你下载的是什么包名字就解压什么名字

4、配置zk1（先配一个节点，然后再复制两份修改相关配置）

改一下解压包名称（便于直观区分）

mv apache-zookeeper-3.6.3 zk1

新建data、logs目录，分别用来存放数据和日志

cd zk1/
mkdir data logs

执行完之后就会看到有 data 和logs 文件夹

进入conf，将zoo_sample.cfg 重命名为 zoo.cfg

cd conf/
cp zoo_sample.cfg /usr/local/zookeeper-cluster/zk1 #可以不执行
mv zoo_sample.cfg zoo.cfg

执行完之后就会看到有 zoo.cfg 可以先复制一份 zoo_sample.cfg zk1文件夹 修改完名称后面在复制进来文件夹里面 怕到时候配置错了 找不到原来的文件

修改conf下的zoo.cfg配置文件

vi zoo.cfg

① 修改：dataDir=/usr/local/zookeeper-cluster/zk1/data

② 添加：dataLogDir=/usr/local/zookeeper-cluster/zk1/logs

③ clientPort=2181【clientPort是客户端的请求端口】

④ 在zoo.cfg文件末尾追加【注意IP是内网IP，伪集群的话写127.0.0.1也可以】

server.1=172.17.80.219:2881:3881

server.2=172.17.80.219:2882:3882

server.3=172.17.80.219:2883:3883

解释：server.实例ID=实例IP:zk服务之间通信端口:zk服务之间投票选举端口；

在 zk1 的 data 目录下创建一个 myid 文件，内容为1

cd zk1/data/
echo 1 > myid

这一步一定要执行！！！

至此，zk1节点配置完成，我们复制两份zk2、zk3，再修改zk2、zk3的相关配置就可以了。

5、配置zk2

用zk1复制一份zk2

cd /usr/local/zookeeper-cluster/
cp -r zk1 zk2

修改zk2的zoo.conf文件

dataDir=/usr/local/zookeeper-cluster/zk2/data
dataLogDir=/usr/local/zookeeper-cluster/zk2/logs
clientPort=2182

修改zk2的data目录下的myid文件

cd /usr/local/zookeeper-cluster/zk2/data/
vi myid

将内容修改为2。

6、配置zk3

用zk1复制一份zk3

cd /usr/local/zookeeper-cluster/
cp -r zk1 zk3

修改zk3的zoo.conf文件

dataDir=/usr/local/zookeeper-cluster/zk3/data
dataLogDir=/usr/local/zookeeper-cluster/zk3/logs
clientPort=2183

修改zk3的data目录下的myid文件

 cd /usr/local/zookeeper-cluster/zk3/data/
 vi myid 

将内容修改为3。

7、开放防火墙端口

开放上面配置涉及到的端口：

2181/2182/2183、2881/2882/2883、3881/3882/3883

举例 开放端口

firewall-cmd --zone=public --add-port=2181/tcp --permanent

重启防火墙

firewall-cmd --reload

查看当前所有tcp端口

netstat -ntlp

8、启动集群

 cd  /usr/local/zookeeper-cluster/zk1/bin/
./zkServer.sh start
 
 cd  /usr/local/zookeeper-cluster/zk2/bin/
 ./zkServer.sh start
 
 cd  /usr/local/zookeeper-cluster/zk3/bin/
 ./zkServer.sh start

9、查看启动状态

 cd  /usr/local/zookeeper-cluster/zk1/bin/
 ./zkServer.sh status
 
 cd  /usr/local/zookeeper-cluster/zk2/bin/
 ./zkServer.sh status
 
 cd  /usr/local/zookeeper-cluster/zk3/bin/
 ./zkServer.sh status

可以看到，zk2作为leader，zk1和zk3作为follower，zookeeper集群搭建成功。

三、搭建kafka集群

1、下载安装包

kafka官网下载：http://kafka.apache.org/downloads
我下载的是 kafka_2.12-3.1.0.gz

2、新建一个kafka-cluster目录，将安装包上传到kafka-cluster目录下

cd /usr/local/
mkdir kafka-cluster

3、解压安装包，重命名

tar -zxvf kafka_2.12-3.1.0.gz
mv kafka_2.12-2.4.0 kafka1

4、修改配置文件

cd /usr/local/kafka-cluster/kafka1/config/
vi server.properties

※修改如下配置项

# 集群内不同实例的broker.id必须为不重复的数字
broker.id=0
# listeners配置kafka的host和port【同样使用内网IP】
listeners=PLAINTEXT://192.168.10.222:9092
# kafka数据和log的存放目录
log.dirs=/usr/local/kafka-cluster/kafka1/logs
# zookeeper集群的ip和端口，用英文逗号分隔
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
# 在配置文件中添加如下配置，表示允许删除topic
delete.topic.enable=true

5、配置kafka2（将kafka1拷一份，修改相关配置）

cp -r kafka1 kafka2
cd kafka2/config/
vi server.properties

配置跟1一样 我就不截图了 你自己对比1的截图修改

broker.id=1
listeners=PLAINTEXT://127.0.0.1:9093
log.dirs=/usr/local/kafka-cluster/kafka2/logs
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
delete.topic.enable=true 

6、配置kafka3（将kafka1拷一份，修改相关配置）

 cp -r kafka1 kafka3
 cd kafka3/config/
 vi server.properties

broker.id=2
listeners=PLAINTEXT://127.0.0.1:9094
log.dirs=/usr/local/kafka-cluster/kafka3/logs
zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
delete.topic.enable=true

7、启动kafka集群（分别启动各实例）

 cd /usr/local/kafka-cluster/kafka1/bin/                     
 ./kafka-server-start.sh ../config/server.properties 
 
 cd /usr/local/kafka-cluster/kafka2/bin/                     
 ./kafka-server-start.sh ../config/server.properties 
 
 cd /usr/local/kafka-cluster/kafka3/bin/                     
 ./kafka-server-start.sh ../config/server.properties 

开三个窗口 ，每个窗口执行一个kafka实例
可以发现在窗口启动之后是一个阻塞进程，会阻塞当前窗口，我们可以重新打开一个窗口进行接下来的操作，或者在启动kafka的时候使用 -daemon 参数将它声明为守护进程后台运行。

 ./kafka-server-start.sh  -daemon ../config/server.properties

※ 启动的时候可能会报错"Cannot allocate memory"

./kafka-server-start.sh  ../config/server.properties 
Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000000c0000000, 1073741824, 0) failed; error='Cannot allocate memory' (errno=12)
#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map 1073741824 bytes for committing reserved memory.
# An error report file with more information is saved as:
# /usr/local/kafka-cluster/kafka2/bin/hs_err_pid28697.log

这是因为单机上搭建伪集群内存不够导致的，我们可以修改启动脚本，将heap内存改小些，默认为1G，可以改为512M，如果还是不够再修改为256M。

 vi bin/kafka-server-start.sh

我没有报错过 所以你们报错了可以试试修改一下启动的shell脚本 按以上操作

8、开放防火墙端口

9092/9093/9094

怎么开放端口 上面有命令可以翻上去看看

9、通过命令简单体验一下kafka

创建一个topic，我们实现生产者向topic写数据，消费者从topic拿数据。

首先进入到bin目录下，
在打开一个新窗口

# cd /usr/local/kafka-cluster/kafka1/bin 

① 创建一个topic

 ./kafka-topics.sh --bootstrap-server 192.168.10.222:9092  --topic mytopic --create  --partitions 1 --replication-factor 2

 参数解释

192.168.10.222:9092 ZK的服务IP:端口号

 --partitions 1 分区数为1

 --topic mytopic  topic的名字是mytopic 

 --replication-factor 副本，注意，副本的个数应小于服务器总数，例如三台服务器做的kafka集群，那么副本数量最多只能为2，也就是 3 - 1 

这样就成功了

查询topics列表

./kafka-topics.sh --bootstrap-server 192.168.10.222:9092 --list
./kafka-topics.sh --bootstrap-serve 192.168.10.222:9092,192.168.10.222:9092 --list  

查看topic详情

./kafka-topics.sh --bootstrap-server 192.168.10.222:9092 --topic mytopic --describe

修改topic，将topic的分区数修改为3，注意“分区数只能增加不能减少。”

./kafka-topics.sh --bootstrap-server 192.168.10.222:9092 --topic mytopic --alter --partitions 3

删除topic

./kafka-topics.sh --bootstrap-server 192.168.10.222:9092 --topic mytopic --delete

生产消息

./kafka-console-producer.sh --topic quickstart-events --bootstrap-server 192.168.10.222:9092

第一次发送

第二次发送

消费消息

./kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server 192.168.10.222:9092

第一次接收

第二次接收

在创建topic时指定的副本数不能大于可用的集群结点数。创建完成后，我们看一下logs目录，

# cd /usr/local/kafka-cluster/kafka2/logs/

可以看到，logs下有两个目录：mytopic-0、mytopic-1、mytopic-2，这就是我们创建的topic的三个分区（我们定义的分区副本是3，这是分区0和1,2的其中一个副本）