0
0

Spring Kafka消费模式（single, batch）及确认模式（自动、手动）示例

目录

Spring Kafka消费消息的模式分为2种模式（对应spring.kafka.listener.type配置）：

  • single - 每次消费单条记录
  • batch - 批量消费消息列表

且每种模式都分为2种提交已消费消息offset的ack模式：

  • 自动确认
  • 手动确认

接下来依次讲解这两种消费模式及其对应的ack模式的示例配置及代码。

1. 单记录消费listener.type=single

本章节先来讲讲record模式 - 单记录消费，且分为自动确认和手动确认2种方式来提交已消费消息offset。

1.1 单记录消费 - 自动确认

即由Spring Kafak框架按照配置规则自动提交已消费消息offset，无需程序手动编码控制。

需注意如下对应配置：

============ 方式1：定时自动提交[不推荐] =====================

开启自动提交（按周期）已消费offset

spring.kafka.consumer.enable-auto-commit

: true

自动提交已消费offset时间价格（配置enable-auto-commit=true时使用）

spring.kafka.consumer.auto-commit-interval

: 1s

========= 方式2：通过ack-mode设置自动提交[推荐] =============

禁用自动提交（按周期）已消费offset

spring.kafka.consumer.enable-auto-commit

: false

listener类型为单条记录single类型（默认为single单条消费模式）

spring.kafka.listener.type

: single

offset提交模式为record

spring.kafka.listener.ack-mode

: record

注：
关于消费者提交已消费消息offset的相关配置说明：

  • spring.kafka.consumer.enbable-auto-commit - true 自动提交已消费消息offset - auto-commit-interval 设置自动提交间隔- fasle 由程序控制已消费消息offset提交 - spring.kafka.listener.ack-mode 已消费offset提交模式

spring.kafka.listener.ack-mode列表（详细说明参见：SpringKafka - Committing Offsets）
ack-mode模式说明自动提交RECORD
单记录当每一条记录被消费者监听器（ListenerConsumer）处理之后提交✔️BATCH（默认）
批量当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交✔️TIME
超时当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
(通过spring.kafka.listener.ack-time设置触发时间)✔️COUNT
超过消费数量当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
(通过spring.kafka.listener.ack-count设置触发数量)✔️COUNT_TIME
超时或超数量TIME或COUNT　有一个条件满足时提交✔️MANUAL
手动提交（ack）后同BATCH当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交❌
需要手动使用
Acknowledgment参数提交MANUAL_IMMEDIATE
手动立即提交手动调用Acknowledgment.acknowledge()后立即提交❌
需要手动使用
Acknowledgment参数提交
**消费端配置示例1 - 定时自动提交

[不推荐]

**

spring:kafka:# 逗号分隔的集群broker列表bootstrap-servers: localhost:9092# ====================================================# ================== 消费者配置 ========================# ====================================================consumer:# 自动提交（按周期）已消费offsetenable-auto-commit:true# 自动提交已消费offset时间价格（配置enable-auto-commit=true时使用）auto-commit-interval: 1s
      ...# ====================================================# ============= 消费者监听器（及线程池）配置 ==============# ====================================================listener:# listener类型# single | batchtype: single
    # ====================================================# ============= 具体业务Kafka定义=======================# ====================================================biz1:topic: topic1
      consumer:group: group1

**消费端配置示例2 - 通过ack-mode设置自动提交

[推荐]

**
listener自动提交offsetd的

ack-mode

模式包括：

RECORD | BATCH | TIME | COUNT | COUNT_TIME

，
且使用相关自动模式

不可

在@KafkaListener标注方法中使用

Acknowledgment

参数。

spring:kafka:# 逗号分隔的集群broker列表bootstrap-servers: localhost:9092# ====================================================# ================== 消费者配置 ========================# ====================================================consumer:# 禁用自动提交（按周期）已消费offsetenable-auto-commit:false...# ====================================================# ============= 消费者监听器（及线程池）配置 ==============# ====================================================listener:# listener类型# single | batchtype: single
      # 已消费offset提交模式（仅在enable-auto-commit=false时才需明确指定）# 单记录  | 批量         | 超时  | 超过消费数量 | 超时或超过数量 | 手动提交（ack）后同BATCH | 手动立即提交# RECORD | BATCH（默认） | TIME | COUNT      | COUNT_TIME   | MANUAL                | MANUAL_IMMEDIATE# https://docs.spring.io/spring-kafka/docs/current/reference/html/#committing-offsets# 注：listener自动提交offset模式包括：RECORD | BATCH | TIME | COUNT | COUNT_TIME ，#    且使用相关自动模式不可在@KafkaListener标注方法中使用Acknowledgment参数ack-mode: record
    # ====================================================# ============= 具体业务Kafka定义=======================# ====================================================biz1:topic: topic1
      consumer:group: group1

消费端代码示例

/**
 * 定义biz1消息接收者
 *
 * @param message
 * @rabbit.exhange exchange1
 * @rabbit.aueue queue1
 * @rabbit.bindingKey #
 */@KafkaListener(
        id ="biz1-${spring.kafka.biz1.consumer.group}",
        groupId ="${spring.kafka.biz1.consumer.group}",
        topics ="${spring.kafka.biz1.topic}")publicvoidbiz1Consumer(String message){
    log.info("[biz1Consumer] RECV MSG: {}", message);}

1.2 单记录消费 - 手动确认

需注意如下对应配置：

禁用自动提交（按周期）已消费offset

spring.kafka.consumer.enable-auto-commit

: false

listener类型为单条记录single类型（默认为single单条消费模式）

spring.kafka.listener.type

: single

offset提交模式为manual_immediate

spring.kafka.listener.ack-mode

: manual_immediate

消费端配置示例
手动提交offset的

ack-mode

模式包括：

MANUAL | MANUAL_IMMEDIATE

，
且使用相关手动模式需在@KafkaListener标注方法中使用

Acknowledgment

参数。

spring:#  profiles:#    active: kafka-originkafka:# 逗号分隔的集群broker列表bootstrap-servers: localhost:9092# ====================================================# ================== 消费者配置 ========================# ====================================================consumer:# 禁用自动提交（按周期）已消费offsetenable-auto-commit:false...# ====================================================# ============= 消费者监听器（及线程池）配置 ==============# ====================================================listener:# listener类型# single | batchtype: single
      # 已消费offset提交模式（仅在enable-auto-commit=false时才需明确指定）# 单记录  | 批量   | 超时  | 超过消费数量 | 超时或超过数量 | 手动提交（ack）后同BATCH | 手动立即提交# RECORD | BATCH | TIME | COUNT      | COUNT_TIME   | MANUAL                | MANUAL_IMMEDIATE# https://docs.spring.io/spring-kafka/docs/current/reference/html/#committing-offsets# 注：手动提交offset模式包括：MANUAL | MANUAL_IMMEDIATE#    且使用相关手动模式需在@KafkaListener标注方法中使用Acknowledgment参数ack-mode: manual_immediate

消费端代码示例
参考：SpringKafka - Manual Acknowledgment

/**
 * 定义biz1消息接收者
 *
 * @param message
 * @kafka.topic topic1
 * @kafka.group group1
 */@KafkaListener(
        id ="biz1-${spring.kafka.biz1.consumer.group}",
        groupId ="${spring.kafka.biz1.consumer.group}",
        topics ="${spring.kafka.biz1.topic}")publicvoidbiz1Consumer(String message, Acknowledgment ack){
    log.info("[biz1Consumer] RECV MSG: {}", message);//确认单当前消息（及之前的消息）offset均已被消费完成
    ack.acknowledge();//拒绝当前消息（此方法仅适用于listener.type=single）//当前poll查询出的剩余消息记录均被抛弃，//且当前消费线程在阻塞指定sleep（如下3000毫秒）后重新调用poll获取待消费消息（包括之前poll被抛弃的消息）//ack.nack(3000)}

2. 批量消费listener.type=batch

在一些需要通过批量处理消息的场景中，SpringKafka支持使用Batch Listeners，即批量处理消息列表。
本章节主要讲解batch模式 - 批量消费，且同样分为自动确认和手动确认2种方式来提交已消费消息offset。

2.1 批量消费 - 自动确认

需注意如下对应配置：

禁用自动提交（按周期）已消费offset

spring.kafka.consumer.enable-auto-commit

: false

批量消费的单次最大消费记录数

spring.kafka.consumer.max-poll-reocrds

: 50

listener类型为批量batch类型（默认为single单条消费模式）

spring.kafka.listener.type

: batch

offset提交模式为batch（不可使用record - 启动报错）

spring.kafka.listener.ack-mode

: batch

配置示例如下

spring:kafka:# 逗号分隔的集群broker列表bootstrap-servers: localhost:9092# ====================================================# ================== 消费者配置 ========================# ====================================================consumer:# 禁用自动提交（按周期）已消费offsetenable-auto-commit:false# 单次poll()调用返回的记录数max-poll-records:50# ====================================================# ============= 消费者监听器（及线程池）配置 ==============# ====================================================listener:# listener类型# single | batchtype: batch
      # 已消费offset提交模式（仅在enable-auto-commit=false时才需明确指定）# 单记录  | 批量   | 超时  | 超过消费数量 | 超时或超过数量 | 手动提交（ack）后同BATCH | 手动立即提交# RECORD | BATCH | TIME | COUNT      | COUNT_TIME   | MANUAL                | MANUAL_IMMEDIATE# https://docs.spring.io/spring-kafka/docs/current/reference/html/#committing-offsetsack-mode: batch
    # ====================================================# ============= 具体业务Kafka定义=======================# ====================================================biz1:topic: topic1
      consumer:group: group1
    biz2:topic: topic2
      consumer:group: group2
        # 分区格式示例：0 | 0,1,2 | 0-3 partitions:0

消费端代码示例

/**
 * 定义biz1消息接收者
 * 自动模式（无需手动ack）：
 * 1. listener.type=batch
 * 2. ack-mode=batch
 *
 * @param messages
 * @kafka.topic topic1
 * @kafka.group group1
 */@KafkaListener(
        id ="biz1-${spring.kafka.biz1.consumer.group}",
        groupId ="${spring.kafka.biz1.consumer.group}",
        topics ="${spring.kafka.biz1.topic}")publicvoidbiz1Consumer(List<String> messages){
    log.info("[biz1Consumer] RECV MSG COUNT: {}", messages.size());
    log.info("[biz1Consumer] RECV MSG[0]: {}", messages.get(0));}/**
 * 定义biz2消息接收者
 * 自动模式（无需手动ack）：
 * 1. listener.type=batch
 * 2. ack-mode=batch
 * 
 * @param messages
 * @kafka.topic topic2
 * @kafka.group group2
 */@KafkaListener(
        id ="biz2-${spring.kafka.biz2.consumer.group}",
        groupId ="${spring.kafka.biz2.consumer.group}",//消费指定分区
        topicPartitions ={@TopicPartition(topic ="${spring.kafka.biz2.topic}", partitions ="${spring.kafka.biz2.consumer.partitions}")})publicvoidbiz2Consumer(List<Message> messages){
    log.info("[biz2Consumer] RECV MSG COUNT: {}", messages.size());
    log.info("[biz2Consumer] RECV MSG[0]: {}", messages.get(0));}

在这里插入图片描述

2.2 批量消费 - 手动确认

需注意如下对应配置：

禁用自动提交（按周期）已消费offset

spring.kafka.consumer.enable-auto-commit

: false

批量消费的单次最大消费记录数

spring.kafka.consumer.max-poll-reocrds

: 50

listener类型为批量batch类型（默认为single单条消费模式）

spring.kafka.listener.type

: batch

offset提交模式为batch（不可使用record - 启动报错）

spring.kafka.listener.ack-mode

: manual

配置示例如下

spring:kafka:# 逗号分隔的集群broker列表bootstrap-servers: localhost:9092# ====================================================# ================== 消费者配置 ========================# ====================================================consumer:# 禁用自动提交（按周期）已消费offsetenable-auto-commit:false# 单次poll()调用返回的记录数max-poll-records:50# ====================================================# ============= 消费者监听器（及线程池）配置 ==============# ====================================================listener:# listener类型# single | batchtype: batch
      # 已消费offset提交模式（仅在enable-auto-commit=false时才需明确指定）# 单记录  | 批量   | 超时  | 超过消费数量 | 超时或超过数量 | 手动提交（ack）后同BATCH | 手动立即提交# RECORD | BATCH | TIME | COUNT      | COUNT_TIME   | MANUAL                | MANUAL_IMMEDIATE# https://docs.spring.io/spring-kafka/docs/current/reference/html/#committing-offsetsack-mode: manual
    # ====================================================# ============= 具体业务Kafka定义=======================# ====================================================biz1:topic: topic1
      consumer:group: group1
    biz2:topic: topic2
      consumer:group: group2
        partitions:0

消费端代码示例

/**
 * 定义biz1消息接收者
 * 手动模式（需手动ack）：
 * 1. listener.type=batch
 * 2. ack-mode=manual
 *
 * @param messages
 * @kafka.topic topic1
 * @kafka.group group1
 */@KafkaListener(
        id ="biz1-${spring.kafka.biz1.consumer.group}",
        groupId ="${spring.kafka.biz1.consumer.group}",//仅在多partition单个消费者时，用于多线程消费消息（concurrency <= partition数量）//当存在多个消费者时，即便设置concurrency > 1也仅有唯一消费线程生效
        concurrency ="${spring.kafka.biz1.consumer.concurrency}",
        topics ="${spring.kafka.biz1.topic}")publicvoidbiz1Consumer(List<String> messages, Acknowledgment ack){
    log.info("[biz1Consumer] RECV MSG COUNT: {}", messages.size());
    log.info("[biz1Consumer] RECV MSG[0]: {}", messages.get(0));//确认单当前消息（及之前的消息）offset均已被消费完成
    ack.acknowledge();//拒绝消息列表中指定index（发生错误的消息index）对应的消息（此方法仅适用于listener.type=batch），//当前指定index之前的消息会被成功提交，//当前poll查询出的剩余消息记录（包括当前指定的index）均被抛弃，//且当前消费线程在阻塞指定sleep（如下3000毫秒）后重新调用poll获取待消费消息（包括当前index及之前poll抛弃的消息）//如下即确认当前list中前5条消息（0-4），抛弃当前list中后续消息，3秒后再次poll查询未消费消息//ack.nack(5, 3000);}/**
 * 定义biz2消息接收者
 * 手动模式（需手动ack）：
 * 1. listener.type=batch
 * 2. ack-mode=manual
 * 
 * @param messages
 * @kafka.topic topic2
 * @kafka.group group2
 */@KafkaListener(
        id ="biz2-${spring.kafka.biz2.consumer.group}",
        groupId ="${spring.kafka.biz2.consumer.group}",//消费指定分区
        topicPartitions ={@TopicPartition(topic ="${spring.kafka.biz2.topic}", partitions ="${spring.kafka.biz2.consumer.partitions}")})publicvoidbiz2Consumer(List<Message> messages, Acknowledgment ack){
    log.info("[biz2Consumer] RECV MSG COUNT: {}", messages.size());
    log.info("[biz2Consumer] RECV MSG[0]: {}", messages.get(0));//确认单当前消息（及之前的消息）offset均已被消费完成
    ack.acknowledge();}

3. 手动模式下的acknowledge和nack方法

在手动确认模式下，除了支持

ack.acknowledge()

方法用于

确认单条记录

（对应

record

模式）或者

批次记录

（对应

batch

模式），
还支持

nack

方法用于拒绝消息，关于

acknowledge

和

nack

方法的详细使用见下表：
方法说明适用listener.typeacknowledge()确认单条消息 或 整批消息single
batchnack(sleepMills)拒绝确认当前消息，
当前poll查询出的剩余消息记录均被抛弃，
且当前消费线程在阻塞指定sleepMills时间后
会重新调用poll获取待消费消息（包括之前poll被抛弃的消息）singlenack(index, sleepMills)拒绝消息列表中指定index（发生错误的消息index）及其之后index对应的消息，
当前指定index之前的消息会被成功提交，
当前poll查询出的剩余消息记录（包括当前指定的index）均被抛弃，
且当前消费线程在阻塞指定sleepMills时间后
会重新调用poll获取待消费消息（包括当前index及之前poll抛弃的消息）batch

标签： kafka spring 消息队列
本文转载自: https://blog.csdn.net/luo15242208310/article/details/122285689
版权归原作者 罗小爬EX 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Spring Kafka消费模式（single, batch）及确认模式（自动、手动）示例”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

机器人技术在安全领域的重要作用

安全学习记录——网络篇（一）

图腾柱PFC：HP1010为您的电动两轮车之旅提供绿色，高效，安全的动力

单元测试数据库回滚问题

这才是单元测试，之前我们都错了

【C语言基础】：字符函数和字符串函数

ChatGPT-Next-Web SSRF漏洞+XSS漏洞复现(CVE-2023-49785)

文章导航