kafka消费的模式及消息积压处理方案

目录

1、kafka消费的流程

2、kafka的消费模式

2.1、点对点模式

2.2、发布-订阅模式

3、consumer消息积压

3.1、处理方案

3.2、积压量

4、消息过期失效

5、kafka注意事项

Kafka消费积压(Consumer Lag)是指消费者处理消息的速度跟不上生产者发送消息的速度，导致消息在Kafka主题中堆积。

关于kakfa的架构图，如下所示：

更多关于kafka的介绍，参考：关于MQ之kafka的深入研究-CSDN博客https://blog.csdn.net/weixin_50055999/article/details/148535599?spm=1011.2415.3001.5331

1、kafka消费的流程

之前的章节中，介绍了kafka消息由producer通过hash函数存放到broker节点后，每个broker节点由多个topic主题组成，可水平扩展。

每个topic由多个partitin组成，partition里面的内容有顺序，跨partition无序。

对于点对点模式下：

消费组内每个消费者可以消费多个partition、同时保留offset偏移位置，保证下次消费。

对于发布订阅模式：

不同消费组内的消费者可以消费同一个patition，两个消费组不受影响，各自保留彼此的offset的偏移位置。

如图所示：

在消费者消费过程的流程如下：

由上图可知：

1、每个topic里面包含多个partition。

2、每个partition里面的内容是按顺序分布的。

3、每个消费者可以消费多个partition。

4、而partition只能被一个消费者消费。

对于不同消费者组，可以共同消费同一个topic里面的消息。

2、kafka的消费模式

Kafka 的消费订阅模式取决于消费者组的配置方式，可以分为以下两种主要模式：

2.1、点对点模式

特点：一条消息只能被一个消费者消费

实现方式：

所有消费者属于同一个消费者组（相同的 group.id）
Kafka 会在组内消费者之间自动平衡分区分配

// 消费者1和消费者2使用相同的group.id
Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("group.id", "my-consumer-group"); // 相同的组ID
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

工作流程：

假设主题有3个分区（P0, P1, P2）
如果有1个消费者，它将消费所有3个分区
如果增加第二个消费者，Kafka会重新平衡：
- 消费者1可能获得P0和P1
- 消费者2获得P2
消息在每个分区内有序，且只被分配给该分区的消费者消费

2.2、发布-订阅模式

特点：一条消息可以被多个消费者（不同消费组）消费（本质还是点对点）

实现方式：

不同消费者组订阅同一个主题
每个消费者组都会收到完整的消息流

// 组A的消费者
Properties propsA = new Properties();
propsA.put("group.id", "group-a"); // 不同组ID
// ...其他配置
KafkaConsumer<String, String> consumerA = new KafkaConsumer<>(propsA);// 组B的消费者
Properties propsB = new Properties();
propsB.put("group.id", "group-b"); // 不同组ID
// ...其他配置
KafkaConsumer<String, String> consumerB = new KafkaConsumer<>(propsB);

工作流程：

生产者发送消息到主题
组A的所有消费者（作为一个组）会收到消息的一个副本
组B的所有消费者（作为另一个独立的组）也会收到消息的一个副本
在每个组内部，消息仍然遵循点对点模式（组内只有一个消费者收到）

3、consumer消息积压

Kafka消息积压的问题，核心原因是生产太快、消费太慢，处理速度长期失衡，从而导致消息积压（Lag）的场景，积压到超过队列长度限制，就会出现还未被消费的数据产生丢失的场景。
如果长时间不解决消息积压，可能会引发资源紧张、服务延迟或崩溃等问题。解决消息积压的关键是提高消费者的消费能力，并优化Kafka集群的整体处理效率。

3.1、处理方案

1. 如果是Kafka消费能力不足，则可以考虑增加 topic 的 partition 的个数（提高kafka的并行度），同时提升消费者组的消费者数量，消费数 = 分区数（二者缺一不可）

2. 若是下游数据处理不及时，则提高每批次拉取的数量。批次拉取数量过少（拉取数据/处理时间 < 生产速度），使处理的数据小于生产的数据，也会造成数据积压。

方法：

1. 增大partion数量。
2. 消费者加了并发，服务，扩大消费线程。
3. 增加消费组服务数量。
4. kafka单机升级成了集群。
5. 避免消费者消费消息时间过长，导致超时。
6. 使Kafka分区之间的数据均匀分布。

3.2、积压量

生产量：Kafka Topic 在一个时间周期内各partition offset 起止时间差值之和。
消费量：Kafka Topic 在一个时间周期内某个消费者的消费量。
积压量：Kafka Topic 的某个Consumer Group残留在消息中间件未被及时消费的消息量。

4、消息过期失效

产生消息堆积，消费不及时，kafka数据有过期时间，一些数据就丢失了，主要是消费不及时。

当出现这种现象的时候，可参考以下经验，进行规避：

1. 消费kafka消息时，应该尽量减少每次消费时间，可通过减少调用三方接口、读库等操作，
从而减少消息堆积的可能性。
2. 如果消息来不及消费，可以先存在数据库中，然后逐条消费（可以保存消费记录，方便定位问题）。
3. 每次接受kafka消息时，先打印出日志，包括消息产生的时间戳。
4. kafka消息保留时间（修改kafka配置文件，默认一周）
5. 任务启动从上次提交offset处开始消费处理

5、kafka注意事项

1. 由于Kafka消息key设置,在Kafka producer处，给key加随机后缀，使其均衡。

2. 数据量很大，合理的增加Kafka分区数是关键。
Kafka分区数是Kafka并行度调优的最小单元，如果Kafka分区数设置的太少，
会影响Kafka consumer消费的吞吐量. 如果利用的是Spark流和Kafka direct approach方式，
也可以对KafkaRDD进行repartition重分区，增加并行度处理.

参考文章：

1、Kafka如何处理大量积压消息_kafka消息堆积过多了怎么办-CSDN博客https://blog.csdn.net/AlbenXie/article/details/128300018?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522dcefb6fbf11572c5ef4526b40c68a37c%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=dcefb6fbf11572c5ef4526b40c68a37c&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~top_click~default-1-128300018-null-null.142^v102^pc_search_result_base1&utm_term=kafka%E6%B6%88%E6%81%AF%E7%A7%AF%E5%8E%8B%E6%80%8E%E4%B9%88%E5%A4%84%E7%90%86&spm=1018.2226.3001.4187