一、安装zookeeper集群
tar -zxvf apache-zookeeper-3.5.5-bin.tar.gz -C /usr/local/ cd /usr/local/ ln -s apache-zookeeper-3.5.5-bin zookeeper mv zoo_sample.cfg zoo.cfg vi zoo.cfg tickTime=2000 initLimit=10 syncLimit=5 dataDir=/tmp/zookeeper/data dataLogDir=/tmp/zookeeper/log clientPort=2181 server.1=192.168.1.12:2888:3888 server.2=192.168.1.13:2888:3888 server.3=192.168.1.14:2888:3888 mkdir /tmp/zookeeper/data -p mkdir /tmp/zookeeper/log -p #在192.168.1.12上 echo 1 >/tmp/zookeeper/data/myid #在192.168.1.13上 echo 2 >/tmp/zookeeper/data/myid #在192.168.1.14上 echo 3 >/tmp/zookeeper/data/myid vi /etc/profile export ZOOKEEPER_HOME=/usr/local/zookeeper export PATH=$PATH:$ZOOKEEPER_HOME/bin source /etc/profile zkServer.sh start zkServer.sh status
二、安装Kafka集群
tar -zxvf kafka_2.11-2.2.1.tgz -C /usr/local/ cd /usr/local/ ln -s kafka_2.11-2.2.1 kafka cd /usr/local/kafka/config vi server.properties broker.id=1 listeners=PLAINTEXT://192.168.1.12:9092 log.dirs=/tmp/kafka-logs #设置可以删除toptic，如果没有设置 delete.topic.enable=true，则调用kafka 的delete命令无法真正将topic删除，而是显示（marked for deletion） delete.topic.enable=true #设置不能自动创建toptic,如果设置为true，则生产者在生产消息时如果toptic不存在则会自动创建 auto.create.topics.enable = false zookeeper.connect=192.168.1.12:2181,192.168.1.12:2181,192.168.1.12:2181 cd /usr/local/kafka/bin ./kafka-server-start.sh -daemon ../config/server.properties
三、Kafka用户权限管理（SASL）和数据传输加密（SSL）
Kafka 权限管理实战（最全整理） https://www.cnblogs.com/felixzh/p/13695298.html Kafka SASL/SCRAM+ACL实现动态创建用户及权限控制(实现用户权限控制) https://blog.csdn.net/ashic/article/details/86661599 Kafka实战：集群SSL加密认证和配置（最新版kafka-2.7.0） https://www.cnblogs.com/felixzh/p/14661156.html kafka学习笔记十kafka-SSL安全认证机制（实现数据安全传输） https://blog.csdn.net/hwhanwan/article/details/82216016 SSL https://blog.csdn.net/qq_38265137/article/details/90112705 https://blog.csdn.net/Tianlujunr/article/details/82379545?utm_medium=distribute.pc_relevant_t0.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-1.control&depth_1-utm_source=distribute.pc_relevant_t0.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-1.control https://www.cnblogs.com/hofmann/p/11174258.html
四、Kafka常用命令
主题topic
a.创建topic
./kafka-topics.sh --create --zookeeper 192.168.1.12:2181 --replication-factor 1 --partitions 1 --topic my-kafka-topic
./kafka-topics.sh --create --bootstrap-server 192.168.1.12:9092 --replication-factor 1 --partitions 1 --topic my-kafka-toptic1 #--bootstrap-server指定broker地址，可以写多个地址，用逗号分割,用法和--zookeeper相同
b.查看topic列表
./kafka-topics.sh --list --zookeeper 192.168.1.13:2181
./kafka-topics.sh --list --bootstrap-server 192.168.1.13:9092
c.如果需要查看topic的详细信息，需要使用describe命令
kafka-topics.sh --describe --zookeeper 192.168.1.12:2181 --topic test-topic
d.#若不指定topic，则查看所有topic的信息
kafka-topics.sh --describe --zookeeper 192.168.1.12:2181
e.删除topic
kafka-topics.sh --delete --zookeeper 192.168.1.12:2181 --topic my-kafka-topic
#https://blog.csdn.net/qq_39657597/article/details/84307541?utm_term=kafka%E5%88%A0%E9%99%A4%E6%B6%88%E6%81%AF&utm_medium=distribute.pc_aggpage_search_result.none-task-blog-2~all~sobaiduweb~default-0-84307541&spm=3001.4430
-------------------------------------------
生产者
kafka-console-producer.sh --broker-list 192.168.1.12:9092 --topic my-kafka-topic
-------------------------------------------
消费者
kafka-console-consumer.sh --bootstrap-server 192.168.1.12:9092 --topic my-kafka-topic
# 如果需要从头开始接收数据，需要添加--from-beginning参数
kafka-console-consumer.sh --bootstrap-server 192.168.1.12:9092 --from-beginning --topic my-kafka-topic
#指定消费者组
./kafka-console-consumer.sh --bootstrap-server 192.168.1.12:9092 --topic my-kafka-toptic1 --group yingyong1 --from-beginning
查看消费者组
./kafka-consumer-groups.sh --bootstrap-server 192.168.1.12:9092 --list
查看消费者组中消费者消费的分区个数
./kafka-consumer-groups.sh --bootstrap-server 192.168.1.12:9092 --describe --members --group yingyong1
重置offset到指定位置（1、--to-offset 30，2、--to-offset --to-earliest，3、--to-offset --to-current，），前提是没有消费者在连着
./kafka-consumer-groups.sh --bootstrap-server 192.168.1.12:9092 --group yingyong2 --reset-offsets --topic my-kafka-topic5 --to-offset 30 --execute
读取指定toptic中的指定分区中的指定offset后面的2条消息
./kafka-console-consumer.sh --bootstrap-server 192.168.1.12:9092 --topic my-kafka-topic5 --partition 1 --offset 10036 --max-messages 2
查看消费者组消息消费情况（查看消息的堆积情况）
./kafka-consumer-groups.sh --bootstrap-server 192.168.1.12:9092 --describe --group yingyong1
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
my-kafka-topic5 4 25 25 0 - - -
my-kafka-topic5 2 26 26 0 - - -
my-kafka-topic5 0 25 25 0 - - -
my-kafka-topic5 3 26 27 1 - - -
my-kafka-topic5 1 25 25 0 - - -
注：LAG当前分区中消息堆积个数
LOG-END-OFFSET当前分区中最后一条消息的位置
CURRENT-OFFSET消费者组中当前分区已经消费到的位置
同一个toptic，不同的消费者组中LOG-END-OFFSET是一样的，而CURRENT-OFFSET是不一样的。
删除toptic中的消息
a.查看分区和offset
./kafka-consumer-groups.sh --bootstrap-server 192.168.1.12:9092 --describe --group yingyong1
b.编写一个json文件
vi /tmp/offsetfile.json
{"partitions": [{"topic": "my-kafka-topic5", "partition": 4, "offset": 27}], "version":1 }
c.执行命令删除
./kafka-delete-records.sh --bootstrap-server 192.168.1.12:9092 --offset-json-file /tmp/offsetfile.json
#########################################
./kafka-topics.sh --describe --zookeeper 192.168.1.12:2181
Topic:my-kafka-topic5 PartitionCount:5 ReplicationFactor:2 Configs:
Topic: my-kafka-topic5 Partition: 0 Leader: 3 Replicas: 2,3 Isr: 3,2
Topic: my-kafka-topic5 Partition: 1 Leader: 3 Replicas: 3,1 Isr: 3,1
Topic: my-kafka-topic5 Partition: 2 Leader: 1 Replicas: 1,2 Isr: 1,2
Topic: my-kafka-topic5 Partition: 3 Leader: 1 Replicas: 2,1 Isr: 1,2
Topic: my-kafka-topic5 Partition: 4 Leader: 3 Replicas: 3,2 Isr: 3,2
注：Isr(in-sync Replica)在同步的副本，Replicas指副本在哪个代理上
一个分区只能被一个消费者组中的一个消费者消费，
如果一个消费者组中有5个分区一个消费者，那么这个消费者会消费这五个分区
如果一个消费者组中有5个分区2个个消费者，那么第一个消费者会消费3个分区，另一个消费者会消费2个分区
如果一个消费者组中有5个分区6个个消费者，那么其中5个消费者会分别消费一个，而第六个消费者不会消费数据，当前五个中有退出的时候，第六个消费者才会消费数据
生产者和消费者在连接kafka时，既可以连接kafka服务器的地址，也可以连接zookeeper的地址，且地址可以写多个中间用逗号隔开即可