kafka2.7.0集群搭建_kafka 2.7

1.准备三台已搭好 zookeeper 集群的主机:01,02,03

2.下载 kafka_2.13-2.7.0.tgz : https://www.apache.org/dyn/closer.cgi?path=/kafka/2.7.0/kafka_2.13-2.7.0.tgz  上传到三台服务

3.解压进入01主机 config 目录修改配置文件 server.properties

broker.id : 在集群中唯一，要求是正数

 

log.dirs : kafka数据的存放地址，多个地址的话用逗号分割,多个目录分布在不同磁盘上可以提高读写性能

三台主机都需要创建 data 目录

 

zookeeper.connect : zookeeper 集群地址,可用ip或者主机名(需要配置hosts,阿里云服务器hosts需要配置内网地址),默认端口 2181 可以不写

 

listeners : 服务器要监听的地址及端口,配置内网地址,没有配置会报错 :

Error while fetching metadata with correlation id 20 : {up_tours=LEADER_NOT_AVAILABLE}

 

将配置文件发送到 02,03 主机,修改 broker.id

4.配置环境变量

5.启动三台 kafka 的 broker :

kafka-server-start.sh /qiouou/kafka/kafka_2.13-2.7.0/config/server.properties

后台启动指定日志

nohup kafka-server-start.sh /qiouou/kafka/kafka_2.13-2.7.0/config/server.properties 2>/dev/null 1>/dev/null &

 

停止 :

6.进入 zookeeper 控制台 : zkCli.sh

查看 kafka 集群

ls /brokers/ids

7.创建主题 topic

在任意一台主机上执行 :

kafka-topics.sh --create --zookeeper qiouou01:2181,qiouou02:2181,qiouou03:2181 --topic up_tours --partitions 3 --replication-factor 3

 

1.一个 broker 服务下,一个主题可以创建多个分区 (--partitions)

2.创建主题时,副本因子应 <= 可用 broker 数 (--replication-factor)

3.消息最大字节数 (--config max.message.bytes=64000)

4.多少条消息来了以后 flush 一次 (--config flush.messages) 

 

8.查看主题列表 : kafka-topics.sh --list --zookeeper qiouou01,qiouou02,qiouou03

9.查看主题详情列表 : kafka-topics.sh --describe --zookeeper localhost:2181

Topic : up_tours        PartitionCount:3        ReplicationFactor:1        Configs:max.message.bytes=64000,flush.messages=1

分区 :

Topic: up_tours        Partition(分区编号): 0        Leader(leader 所在 broker 的 id ): 3        Replicas(所有副本列表): 3,2,1        Isr(可用副本列表): 3,2,1

Topic: up_tours        Partition(分区编号): 1        Leader(leader 所在 broker 的 id ): 1        Replicas(所有副本列表): 1,3,2        Isr(可用副本列表): 1

Topic: up_tours        Partition(分区编号): 2        Leader(leader 所在 broker 的 id ): 2        Replicas(所有副本列表): 2,1,3        Isr(可用副本列表): 2,1,3

10.查看单个主题 : kafka-topics.sh --describe --zookeeper qiouou01,qiouou02,qiouou03 --topic up_tours

11.查看主题是否存在 : kafka-topics.sh --list --zookeeper qiouou01,qiouou02,qiouou03|grep mytopic2 

12.修改主题分区数 : kafka-topics.sh --alter --zookeeper qiouou01,qiouou02,qiouou03 --topic mytopic --partitions 5

13.删除配置 :

kafka-topics.sh --alter --zookeeper qiouou01,qiouou02,qiouou03 --topic mytopic3 --delete-config max.message.bytes  --delete-config flush.messages

14.删除主题 :

kafka-topics.sh --zookeeper qiouou01,qiouou02,qiouou03 --delete --topic mytopic1

只是标记删除,在 Kafka 重启后删除(0.10不会删除),如果需要删除 需在 server.properties 中配置 delete.topic.enable=true

重启还是有主题,需要去zookeeper中删除主题
1.进入zookeeper客户端 : zkCli.sh

2.查看主题列表: ls /borkers/topics

3.1 rmr 命令删除 rmr /brokers/topics/topicname (可能找不到 rmr 命令)

3.2 deleteall  /brokers/topics/topicname

 

15.启动生产者 : kafka-console-producer.sh --topic up_tours --broker-list  qiouou01:9092,qiouou02:9092,qiouou03:9092

16.创建消费者 :

kafka-console-consumer.sh --bootstrap-server qiouou01:9092,qiouou02:9092,qiouou03:9092 --topic up_tours --consumer-property group.id=group_1

--consumer-property group.id : 指定消费组 ID

 

17.创建消费者改变消费组ID,创建好过后之前的消息 1,hello,不会出现,因为发消息之前没有订阅,现在订阅了,停止消费组3,再生产消息,再启动消费组3

kafka-console-consumer.sh --bootstrap-server qiouou01:9092,qiouou02:9092,qiouou03:9092 --topic up_tours --consumer-property group.id=group_2  

生产者

消费者组3

18.要点

1.单个分区中的数据时有序的,且不可修改,但不同分区中数据是无序的

2.同一组的消费者对于同一条消息只能消费一次

3.一个消费组中的消费者应该小于等于一个主题下的分区数(最好是分区数的整倍数)

partition 数量决定了每个 consumer group 中并发消费者的最大数量

4.同一分区,同一时刻只能一个消费组中的一个消费者消费

5.offset 记录消息数,记录消费进度,指定消息消费(对同一条消息可以消费多次)

 

副本因子

1.副本因子小于等于可用 broker 数

2.以分区为副本

3.让 Kafka 高可靠

4.同一个副本不能放在同一个 broker 中

5.只有一个 leader 对外开放

6.如果一个分区下有三个副本,其中一个挂掉,会在剩下两个中选取 leader ,不会在其他 broker 中重启一个副本

   因为主机之间传递数据会长时间占用网络 io,会影响 kafka 的高吞吐量

7.如果所有副本都挂了,生产者指定了分区的话,将写入不成功,消息是待写入,等待服务重启