kafka+zk集群部署环境要求

jdk 1.8.0_211

apache-zookeeper-3.5.7-bin

kafka_2.13-2.6.0

服务器要求

系统：centos 7.6

服务器数量：三台

T1 192.168.0.5

T2 192.168.0.6

T3 192.168.0.7

Zookeeper集群部署增加hosts配置：

vi /etc/hosts

192.168.0.5 T1

192.168.0.6 T2

192.168.0.7 T3

下载安装包: 解压安装：

tar -xzf apache-zookeeper-3.5.7-bin.tar.gz -C /usr/local/

ln -s /usr/local/apache-zookeeper-3.5.7-bin/ /usr/local/zookeeper

增加配置：

vi /usr/local/zookeeper/conf/zoo.cfg

#确保zookeeper存储目录存在

#mkdir –p /home/data/zookeeper/; mkdir –p /home/logs/zookeeper/

tickTime=2000

initLimit=10

syncLimit=5

dataDir=/home/data/zookeeper/

dataLogDir=/home/logs/zookeeper/

clientPort=2181

#maxClientCnxns=60

server.1=192.168.0.5:2888:3888

server.2=192.168.0.6:2888:3888

server.3=192.168.0.7:2888:3888

创建myid

这个步骤需要分别在T1,T2,T3创建不同的id来区分集群的机器，而且这个id必须跟上面zoo.cfg中server.后面的数字对应。

T1执行

echo '1' > /home/data/zookeeper/myid

T2执行

echo '2' > /home/data/zookeeper/myid

T3执行

echo '3' > /home/data/zookeeper/myid

启动服务

启动：/usr/local/zookeeper/bin/zkServer.sh start

停止：/usr/local/zookeeper/bin/zkServer.sh stop

查看状态: /usr/local/zookeeper/bin/zkServer.sh status

Kafka集群部署解压安装

tar -xzf kafka_2.13-2.6.0.tgz -C /usr/local/

ln -s /usr/local/kafka_2.13-2.6.0/ /usr/local/kafka

修改配置：

创建目录: mkdir /home/logs/kafka-logs

清空配置: >/usr/local/kafka/config/server.properties

增加配置: vi /usr/local/kafka/config/server.properties

#注意broker id 三台服务器不能一致，跟zookeeper id对应1 2 3

broker.id=1

#listeners=PLAINTEXT://:9092

#advertised.listeners=PLAINTEXT://your.host.name:9092

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

log.dirs=/home/logs/kafka-logs

num.partitions=1

num.recovery.threads.per.data.dir=1

offsets.topic.replication.factor=1

transaction.state.log.replication.factor=1

transaction.state.log.min.isr=1

log.retention.hours=720

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

#zookeeper集群地址

zookeeper.connect=192.168.0.5:2181,192.168.0.6:2181,192.168.0.7:2181

zookeeper.connection.timeout.ms=18000

group.initial.rebalance.delay.ms=0

offsets.commit.required.acks = -1

#副本数

offsets.topic.replication.factor=3

transaction.state.log.replication.factor=3

启动服务

启动：

/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

停止：

/usr/local/kafka/bin/kafka-server-stop.sh

查看集群状态:

/usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper T1

/usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper T2

/usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper T3

Kafka测试创建 Topic

参数 --topic 指定 Topic 名，--partitions 指定分区数，--replication-factor 指定备份数：

/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper T1:2181,T2:2181,T3:2181 --replication-factor 3 --partitions 1 --topic test

消费消息

在T1服务器执行

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server T1:9092,T2:9092,T3:9092 --topic test --from-beginning

停止节点

在Leader节点服务器执行，Leader使用查看集群状态命令查看

例如：

Leader为2，在T2服务器执行

/usr/local/kafka/bin/kafka-server-stop.sh

查看集群状态:

/usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper 192.168.0.6

生产消息

在T3服务器执行

/usr/local/kafka/bin/kafka-console-producer.sh --broker-list T1:9092,T2:9092,T3:9092 --topic test

六、kafka基本使用说明

Kafka

# Start the ZooKeeper service

# Note: Soon, ZooKeeper will no longer be required by Apache Kafka.

$ bin/zookeeper-server-start.sh config/zookeeper.properties

# Start the Kafka broker service

$ bin/kafka-server-start.sh config/server.properties

启动 Kafka

后台常驻方式，带上参数 -daemon，如：

/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

指定 JMX port 端口启动，指定 jmx，可以方便监控 Kafka 集群

JMX_PORT=9991 /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

停止 Kafka

/usr/local/kafka/bin/kafka-server-stop.sh

Topic

创建 Topic

参数 --topic 指定 Topic 名，--partitions 指定分区数，--replication-factor 指定备份数：

/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper T1:2181,T2:2181,T3:2181 --replication-factor 3 --partitions 1 --topic test

注意，如果配置文件 server.properties 指定了 Kafka 在 zookeeper 上的目录，则参数也要指定，否则会报无可用的 brokers（下面部分命令也有同样的情况），如：

/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181/kafka --replication-factor 1 --partitions 1 --topic test

列出所有 Topic

/usr/local/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181

查看 Topic

/usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

增加 Topic 的 partition 数

/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic test --partitions 5

查看 topic 指定分区 offset 的最大值或最小值

time 为 -1 时表示最大值，为 -2 时表示最小值：

/usr/local/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic test --time -1 --broker-list 127.0.0.1:9092 --partitions 0

删除 Topic

/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --topic test --delete

生产消息

/usr/local/kafka/bin/kafka-console-producer.sh --broker-list T1:9092,T2:9092,T3:9092 --topic test

消费消息

从头开始

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server T1:9092,T2:9092,T3:9092 --topic test --from-beginning

从尾部开始

从尾部开始取数据，必需要指定分区：

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --offset latest --partition 0

指定分区

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --offset latest --partition 0

取指定个数

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --offset latest --partition 0 --max-messages 1

消费者 Group

指定 Group

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test -group test_group --from-beginning

消费者 Group 列表

/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

查看 Group 详情

/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --describe

输出：

Consumer group 'test_group' has no active members.

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID

test 0 5 5 0 - - -

# CURRENT-OFFSET: 当前消费者群组最近提交的 offset，也就是消费者分区里读取的当前位置

# LOG-END-OFFSET: 当前最高水位偏移量，也就是最近一个读取消息的偏移量，同时也是最近一个提交到集群的偏移量

# LAG：消费者的 CURRENT-OFFSET 与 broker 的 LOG-END-OFFSET 之间的差距

删除 Group 中 Topic

/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --topic test --delete

删除 Group

/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test_group --delete

平衡 leader

/usr/local/kafka/bin/kafka-preferred-replica-election.sh --bootstrap-server localhost:9092

自带压测工具

/usr/local/kafka/bin/kafka-producer-perf-test.sh --topic test --num-records 100 --record-size 1 --throughput 100 --producer-props bootstrap.servers=localhost:9092

创建

/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic config_product

/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic matching_order_ADO-USDT

/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic matching_message_ADO-USDT