一、环境准备：
首先准备好三台CentOS系统的主机，设置ip为：172.16.20.220、172.16.20.221、172.16.20.222。
Kafka会使用大量文件和网络socket，Linux默认配置的File descriptors（文件描述符）不能够满足Kafka高吞吐量的要求，所以这里需要调整(更多性能优化，请查看Kafka官方文档):
vi /etc/security/limits.conf
# 在最后加入，修改完成后，重启系统生效。
* soft nofile 131072
* hard nofile 131072
新建kafka的日志目录和zookeeper数据目录，因为这两项默认放在tmp目录，而tmp目录中内容会随重启而丢失,所以我们自定义以下目录:
mkdir /data/zookeeper
mkdir /data/zookeeper/data
mkdir /data/zookeeper/logs
mkdir /data/kafka
mkdir /data/kafka/data
mkdir /data/kafka/logs
二、zookeeper.properties配置
vi /usr/local/kafka/config/zookeeper.properties
修改如下：
# 修改为自定义的zookeeper数据目录
dataDir=/data/zookeeper/data
# 修改为自定义的zookeeper日志目录
dataLogDir=/data/zookeeper/logs
# 端口
clientPort=2181
# 注释掉
#maxClientCnxns=0
# 设置连接参数，添加如下配置
# 为zk的基本时间单元，毫秒
tickTime=2000
# Leader-Follower初始通信时限 tickTime*10
initLimit=10
# Leader-Follower同步通信时限 tickTime*5
syncLimit=5
# 设置broker Id的服务地址，本机ip一定要用0.0.0.0代替
server.1=0.0.0.0:2888:3888
server.2=172.16.20.221:2888:3888
server.3=172.16.20.222:2888:3888
三、在各台服务器的zookeeper数据目录/data/zookeeper/data添加myid文件，写入服务broker.id属性值
在data文件夹中新建myid文件，myid文件的内容为1（一句话创建：echo 1 > myid）
cd /data/zookeeper/data
vi myid
#添加内容：1 其他两台主机分别配置 2和3
1
四、kafka配置，进入config目录下，修改server.properties文件
vi /usr/local/kafka/config/server.properties
# 每台服务器的broker.id都不能相同
broker.id=1
# 是否可以删除topic
delete.topic.enable=true
# topic 在当前broker上的分片个数，与broker保持一致
num.partitions=3
# 每个主机地址不一样：
listeners=PLAINTEXT://172.16.20.220:9092
advertised.listeners=PLAINTEXT://172.16.20.220:9092
# 具体一些参数
log.dirs=/data/kafka/kafka-logs
# 设置zookeeper集群地址与端口如下：
zookeeper.connect=172.16.20.220:2181,172.16.20.221:2181,172.16.20.222:2181
五、Kafka启动
kafka启动时先启动zookeeper，再启动kafka；关闭时相反，先关闭kafka，再关闭zookeeper。
1、zookeeper启动命令
./zookeeper-server-start.sh ../config/zookeeper.properties &
后台运行启动命令：
nohup ./zookeeper-server-start.sh ../config/zookeeper.properties >/data/zookeeper/logs/zookeeper.log 2>1 &
或者
./zookeeper-server-start.sh -daemon ../config/zookeeper.properties &
查看集群状态：
./zookeeper-server-start.sh status ../config/zookeeper.properties
2、kafka启动命令
./kafka-server-start.sh ../config/server.properties &
后台运行启动命令：
nohup bin/kafka-server-start.sh ../config/server.properties >/data/kafka/logs/kafka.log 2>1 &
或者
./kafka-server-start.sh -daemon ../config/server.properties &
3、创建topic，最新版本已经不需要使用zookeeper参数创建。
./kafka-topics.sh --create --replication-factor 2 --partitions 1 --topic test --bootstrap-server 172.16.20.220:9092
参数解释:
复制两份
--replication-factor 2
创建1个分区
--partitions 1
topic 名称
--topic test
4、查看已经存在的topic（三台设备都执行时可以看到）
./kafka-topics.sh --list --bootstrap-server 172.16.20.220:9092
5、启动生产者：
./kafka-console-producer.sh --broker-list 172.16.20.220:9092 --topic test
6、启动消费者：
./kafka-console-consumer.sh --bootstrap-server 172.16.20.221:9092 --topic test
./kafka-console-consumer.sh --bootstrap-server 172.16.20.222:9092 --topic test
添加参数 --from-beginning 从开始位置消费，不是从最新消息
./kafka-console-consumer.sh --bootstrap-server 172.16.20.221 --topic test --from-beginning
7、测试：在生产者输入test，可以在消费者的两台服务器上看到同样的字符test，说明Kafka服务器集群已搭建成功。
网友评论