Kafka 一、介绍与使用

kafka是一个消息中间件，是一个分布式的，支持分区的，多副本的，基于zookeeper的分布式消息系统。

kafka基本概念

• Broker ：kafka集群中的每一个节点称为broker
• Topic : kafka中消息根据topic分类，生产消息和消费消息必须要订阅topic
• Partition :分区。每个topic下可以若干个分区，每个分区内部的消息都是有序的，是真正存储消息的区域
• Producer :生产者，产生消息的客户端
• Consumer :消费者，消费消息的客户端
• GroupComsumer :组消费者，一条消息只能被一个组消费，如果有多个组订阅了相同的topic，每个组内都可以消费，且每个组只能有一个consumer客户端消费

安装并启动

kafka依赖jvm和zookeeper，需要先安装

zookeeper中存放了kafka broker的元数据信息

# 下载解压kafka
wget  https://archive.apache.org/dist/kafka/2.4.1/kafka_2.11-2.4.1.tgz
tar -zxcf kafka_2.11-2.4.1.tgz
#修改配置文件
cd kafka_2.11-2.4.1/config
vim server.properties
# broker.id=0 属性在kafka集群中必须要是唯一
# listeners=PLAINTEXT://192.168.10.102:9092 对外提供的地址+端口号
# log.dir=/data/kafka/kafka-logs kafka的消息存储文件
# zookeeper.connect=192.168.10.102:2181 kafka连接zookeeper的地址

#启动
../bin/kafka-server-start.sh -daemon server.properties

#查看是否启动成功，根据zookeeper中的节点
zkClis.sh
ls /brokers/ids #查看kafka节点
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

Kafka基本使用

创建topic

#指定分区数是1 副本数是1 topic名称是test
./kafka-topics.sh --create --zookeeper 192.168.10.102:2181 --replication-factor 1 --partitions 1 --topic test
1
2

删除topic

#删除名称是test的topic
./kafka-topics.sh --delete --topic test --zookeeper 192.168.10.102:2181
1
2

发送消息

#创建producer,指定topic为test 在默认情况下，每一个行会被当做成一个独立的消息
./kafka-console-producer.sh --broker-list 192.168.10.102:9092 --topic test
> 111
> 222
> 333
1
2
3
4
5

消费消息

# 消费test主题下的消息，默认消费最新的消息
./kafka-console-consumer.sh --bootstrap-server 192.168.10.102:9092 --topic test
# --from-beginning 从头开始消费
./kafka-console-consumer.sh --bootstrap-server 192.168.10.102:9092 --from-beginning --topic test
# ‐‐whitelist 消费多个topic
./kafka-console-consumer.sh --bootstrap-server 192.168.10.102:9092 --whitelist "test|test-2"
1
2
3
4
5
6

以上命令如果开启多个客户端的话，是都可以消费到生产者发来的消息的，默认不指定组的话，就是广播消息。如果我们希望启动多个客户端，只能有一个客户端去消费，就需要指定group

单播消费

# 启动两个客户端，执行相同的命令，只有一个客户端消费信息
./kafka-console-consumer.sh --bootstrap-server 192.168.10.102:9092 --consumer-property group.id=testGroup  --topic test
1
2

多播消费

# 启动两个客户端，分别使用不同的组去消费，都可以消费消息
./kafka-console-consumer.sh --bootstrap-server 192.168.10.102:9092 --consumer-property group.id=testGroup1  --topic test
./kafka-console-consumer.sh --bootstrap-server 192.168.10.102:9092 --consumer-property group.id=testGroup2  --topic test
1
2
3

生产者发送消息

> 111
> 222 
#断掉testGroup2消费组的客户端
> 333
> 444
1
2
3
4
5

查看消费组的列表

# 查询消费组列表
./kafka-consumer-groups.sh --bootstrap-server 192.168.10.102:9092 --list
# testGroup2
# testGroup1
# testGroup
1
2
3
4
5

分别查询消费组的状态

./kafka-consumer-groups.sh --bootstrap-server 192.168.10.102:9092 --describe --group testGroup1
./kafka-consumer-groups.sh --bootstrap-server 192.168.10.102:9092 --describe --group testGroup2
1
2

• current-offset：当前消费组的已消费偏移量
• log-end-offset：主题对应分区消息的结束偏移量(HW)
• lag：当前消费组未消费的消息数

当group客户端断掉重启之后，默认会从current-offset开始消费

---

上面提到一个topic下可以有多个partition

partition是一个有序的message序列，会按照顺序添加到commit log文件中，每个partition中的消息都会有个唯一的编号offset。一个partition中的offset值是唯一的，不同的partition的offset可能会相同。所以consumer可以指定offset重复消费消息或者跳过某些消息。

kafka一般不会删除消息，不管这些消息有没有被消费。只会根据配置的日志保留时间(log.retention.hours)确认消息多久被删除，默认保留最近一周的日志消息。kafka的性能与保留的消息数据量大小没有关系，因此保存大量的数据消息日志信息不会有什么影响。

创建多分区主题

./kafka-topics.sh --create --zookeeper 192.168.10.102:2181 --replication-factor 1 --partitions 2 --topic test2
#查看topic的状态
./kafka-topics.sh --describe --zookeeper 192.168.10.102:2181  --topic test2

1
2
3
4

可以看到有两个partition，其他一些信息在集群部署时会用到

• leader：表示哪个broker是leader，负责给定partition的所有读写请求
• replicas： 表示在哪些broker机器上有副本
• isr：是replicas的一个子集，它只列出当前还存活着的，并且已同步备份了该partition的broker

在kafka-logs目录下可以看到test2主题的两个分区