Kafka 队列（共享消费者）

Doubao 中英对照 Kafka Queues (Share Consumer)

自 4.0 版本起，Spring for Apache Kafka 通过共享消费者（share consumer）为 Kafka 队列提供支持，实现了 KIP-932（Kafka 队列）。截至 Spring for Apache Kafka 4.1，以 Apache Kafka 4.2 为底座，共享消费者已晋升为完整的生产级支持。

与传统消费者组相比，Kafka 队列启用了一种不同的消费模型。它不是基于分区的分配模型（每个分区被独占分配给一个消费者），共享消费者可以协作地从相同分区消费，记录在共享组中的各消费者之间分发。

共享消费者工厂

ShareConsumerFactory 负责创建共享消费者实例。Spring Kafka 提供了 DefaultShareConsumerFactory 实现。

配置

你可以像配置常规 ConsumerFactory 一样配置一个 DefaultShareConsumerFactory：

@Bean
public ShareConsumerFactory<String, String> shareConsumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-share-group");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return new DefaultShareConsumerFactory<>(props);
}

构造选项

DefaultShareConsumerFactory 提供了若干构造选项：

// Basic configuration
new DefaultShareConsumerFactory<>(configs);

// With deserializer suppliers
new DefaultShareConsumerFactory<>(configs, keyDeserializerSupplier, valueDeserializerSupplier);

// With deserializer instances
new DefaultShareConsumerFactory<>(configs, keyDeserializer, valueDeserializer, configureDeserializers);

反序列化器配置

你可以通过多种方式配置反序列化器：

1. 通过配置属性（简单场景推荐）：
2. 通过 setter：
3. 通过 Supplier（适用于需要为每个消费者创建反序列化器的场景）：

如果你的反序列化器已经完全配置好、不应被工厂重新配置，则将 configureDeserializers 设为 false。

生命周期监听器

你可以添加监听器来监控共享消费者的生命周期：

factory.addListener(new ShareConsumerFactory.Listener<String, String>() {
    @Override
    public void consumerAdded(String id, ShareConsumer<String, String> consumer) {
        // Called when a new consumer is created
        System.out.println("Consumer added: " + id);
    }

    @Override
    public void consumerRemoved(String id, ShareConsumer<String, String> consumer) {
        // Called when a consumer is closed
        System.out.println("Consumer removed: " + id);
    }
});

共享消息监听器容器

ShareKafkaMessageListenerContainer

ShareKafkaMessageListenerContainer 为共享消费者提供了一个支持并发处理的容器：

@Bean
public ShareKafkaMessageListenerContainer<String, String> container(
        ShareConsumerFactory<String, String> shareConsumerFactory) {

    ContainerProperties containerProps = new ContainerProperties("my-topic");
    containerProps.setGroupId("my-share-group");

    ShareKafkaMessageListenerContainer<String, String> container =
        new ShareKafkaMessageListenerContainer<>(shareConsumerFactory, containerProps);

    container.setupMessageListener(new MessageListener<String, String>() {
        @Override
        public void onMessage(ConsumerRecord<String, String> record) {
            System.out.println("Received: " + record.value());
        }
    });

    return container;
}

容器属性

共享容器支持常规消费者可用容器属性的一个子集：

• topics：要订阅的主题名称数组
• groupId：共享组 ID
• clientId：消费者的客户端 ID
• shareAckMode：确认模式（EXPLICIT、MANUAL 或 IMPLICIT；见 记录确认）
• shareAcknowledgmentTimeout：在 MANUAL 模式下检测未确认记录的超时时间
• syncShareCommits：共享消费者确认使用 commitSync()（默认）还是 commitAsync()（见 同步与异步提交）
• kafkaConsumerProperties：额外的消费者属性

important

共享消费者不支持：

• 显式分区分配（TopicPartitionOffset）
• 主题模式
• 手动偏移量管理

生命周期事件

ShareKafkaMessageListenerContainer 会在整个消费者生命周期发布 Spring 应用事件：

• ConsumerStartingEvent：共享消费者线程首次启动时、在它开始轮询之前发布。
• ConsumerStartedEvent：共享消费者即将开始轮询时发布。
• ConsumerFailedToStartEvent：共享消费者启动失败时发布（例如底层 ShareConsumer 无法创建）。同一启动尝试中已构造的任何 ShareConsumer 实例都会被静默关闭，并且不发布停止或已停止事件，因为它们从未到达轮询循环。
• ShareConsumerStoppingEvent：由每个共享消费者在其底层 ShareConsumer 关闭之前发布。与 ConsumerStoppingEvent 不同，此事件携带一个 ShareConsumer 引用（而非 Consumer），因为 ShareConsumer 是一个独立的客户端 API，不分配给特定分区。
• ConsumerStoppedEvent：共享消费者关闭后发布，带有一个 reason：
  • NORMAL——容器被停止。
  • ABNORMAL——消费者线程因 Exception 退出。
  • ERROR——抛出了一个 java.lang.Error。

这些事件可以用 @EventListener 消费（见 事件消费），从而以与常规容器相同的监控和重启模式集成共享容器。

并发

ShareKafkaMessageListenerContainer 通过在单个容器内创建多个消费者线程来支持并发处理。每个线程运行自己的 ShareConsumer 实例，参与同一个共享组。

与传统消费者组（并发涉及分区分配）不同，共享消费者利用 Kafka 在代理端的记录级分发。这意味着同一容器中的多个消费者线程作为共享组的一部分协同工作，由 Kafka 代理把记录分发给所有消费者实例。

important

并发在应用实例间是累加的

从共享组的视角看，每个 ShareConsumer 实例都是一个独立成员，无论它运行在哪里。在单个容器中设置 concurrency=3 会创建 3 个共享组成员。如果你用同一共享组 ID 运行多个应用实例，它们所有的消费者线程合并为一个池。

例如：* 应用实例 1：concurrency=3 → 3 个共享组成员 * 应用实例 2：concurrency=3 → 3 个共享组成员 * 总计：6 个共享组成员可供代理分发记录。

这意味着在单个容器中设置 concurrency=5，在运维上等价于运行 5 个各自 concurrency=1 的独立应用实例（都使用相同的 group.id）。Kafka 代理平等对待所有消费者实例，把记录分发给整个池。

以编程方式配置并发

@Bean
public ShareKafkaMessageListenerContainer<String, String> concurrentContainer(
        ShareConsumerFactory<String, String> shareConsumerFactory) {

    ContainerProperties containerProps = new ContainerProperties("my-topic");
    containerProps.setGroupId("my-share-group");

    ShareKafkaMessageListenerContainer<String, String> container =
        new ShareKafkaMessageListenerContainer<>(shareConsumerFactory, containerProps);

    // Set concurrency to create 5 consumer threads
    container.setConcurrency(5);

    container.setupMessageListener(new MessageListener<String, String>() {
        @Override
        public void onMessage(ConsumerRecord<String, String> record) {
            System.out.println("Received on " + Thread.currentThread().getName() + ": " + record.value());
        }
    });

    return container;
}

通过工厂配置并发

你可以在工厂层设置默认并发，它适用于该工厂创建的所有容器：

@Bean
public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerContainerFactory(
        ShareConsumerFactory<String, String> shareConsumerFactory) {

    ShareKafkaListenerContainerFactory<String, String> factory =
        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);

    // Set default concurrency for all containers created by this factory
    factory.setConcurrency(3);

    return factory;
}

每个监听器的并发

可以使用 concurrency 属性按每个监听器覆盖并发设置：

@Component
public class ConcurrentShareListener {

    @KafkaListener(
        topics = "high-throughput-topic",
        containerFactory = "shareKafkaListenerContainerFactory",
        groupId = "my-share-group",
        concurrency = "10"  // Override factory default
    )
    public void listen(ConsumerRecord<String, String> record) {
        // This listener will use 10 consumer threads
        System.out.println("Processing: " + record.value());
    }
}

并发注意事项

• 线程安全：每个消费者线程有自己的 ShareConsumer 实例，并独立管理自己的确认
• 客户端 ID：每个消费者线程收到一个带数字后缀的唯一客户端 ID（例如 myContainer-0、myContainer-1 等）
• 指标：所有消费者线程的指标被聚合，可通过 container.metrics() 访问
• 生命周期：所有消费者线程作为一个单元一起启动和停止
• 工作分发：Kafka 代理负责把记录分发给共享组中的所有消费者实例
• 确认：每个线程独立管理自己记录的确认；一个线程中未确认的记录不会阻塞其他线程

带 MANUAL 确认的并发

并发与 ShareAckMode.MANUAL 无缝协作。每个消费者线程独立跟踪并确认自己的记录：

@KafkaListener(
    topics = "order-queue",
    containerFactory = "manualShareKafkaListenerContainerFactory",
    groupId = "order-processors",
    concurrency = "5"
)
public void processOrder(ConsumerRecord<String, String> record, ShareAcknowledgment acknowledgment) {
    try {
        processOrderLogic(record.value());
        acknowledgment.acknowledge(); // ACCEPT
    }
    catch (RetryableException e) {
        acknowledgment.release(); // Will be redelivered
    }
    catch (Exception e) {
        acknowledgment.reject(); // Permanent failure
    }
}

备注

记录获取与分发行为：

共享消费者使用拉取模型，每个消费者线程调用 poll() 从代理获取记录。当一个消费者轮询时，代理的共享分区领导者：

• 选择 "Available"（可用）状态的记录
• 把它们转为 "Acquired"（已获取）状态，并附带一个限时获取锁（默认 30 秒，可通过 group.share.record.lock.duration.ms 配置）
• 为效率偏好返回完整的记录批
• 默认将 max.poll.records 作为软限制，意味着即使超出该值也会获取完整的记录批

===== Acquire Mode (KIP-1206)

Apache Kafka 4.2 引入了 share.acquire.mode 消费者属性，用于控制 max.poll.records 的严格程度：

• batch_optimized（默认）——软限制。代理可以返回超过 max.poll.records 的记录，以对齐批边界从而获得更好的吞吐。
• record_limit——严格限制。代理每次 poll() 调用返回的记录数绝不超过 max.poll.records。当你需要对内存用量或处理速率进行精确控制时很有用。

这是一个直通型消费者属性。直接在消费者属性中配置它：

Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put("share.acquire.mode", "record_limit");
props.put("max.poll.records", 5);
// ... other properties

DefaultShareConsumerFactory<String, String> factory = new DefaultShareConsumerFactory<>(props);

备注

无需额外的 Spring Kafka 配置。该属性被直接传给底层的 KafkaShareConsumer。

当记录被一个消费者获取时，其他消费者无法使用它们。当获取锁过期时，未确认的记录自动回到 "Available" 状态，可以投递给另一个消费者。

代理使用 group.share.partition.max.record.locks 限制每个分区可获取的记录数。一旦达到此限制，后续轮询暂时不返回记录，直到锁过期。

对并发的影响：

• 每个消费者线程独立轮询，每次轮询可能获取不同数量的记录
• 跨线程的记录分发取决于轮询时机和批的可用性
• 多线程扩大了可获取记录的消费者池
• 在消息量低或分区少的情况下，记录可能集中在较少的线程上
• 对于长时间运行的工作负载，分发往往更均匀

配置：

• 每个线程独立轮询并处理记录
• 确认约束按线程适用（一个线程的未确认记录不会阻塞其他线程）
• 并发设置必须大于 0，且容器运行时不能更改
  :::

注解驱动的监听器

用共享消费者的 @KafkaListener

你可以通过配置一个 ShareKafkaListenerContainerFactory，将 @KafkaListener 与共享消费者一起使用：

@Configuration
@EnableKafka
public class ShareConsumerConfig {

    @Bean
    public ShareConsumerFactory<String, String> shareConsumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultShareConsumerFactory<>(props);
    }

    @Bean
    public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerContainerFactory(
            ShareConsumerFactory<String, String> shareConsumerFactory) {
        return new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
    }
}

然后在你的监听器中使用它：

@Component
public class ShareMessageListener {

    @KafkaListener(
        topics = "my-queue-topic",
        containerFactory = "shareKafkaListenerContainerFactory",
        groupId = "my-share-group"
    )
    public void listen(ConsumerRecord<String, String> record) {
        System.out.println("Received from queue: " + record.value());
        // Record is automatically acknowledged with ACCEPT
    }
}

共享组偏移量重置

与常规消费者组不同，共享组使用不同的偏移量重置行为配置。你可以以编程方式配置它：

private void configureShareGroup(String bootstrapServers, String groupId) throws Exception {
    Map<String, Object> adminProps = new HashMap<>();
    adminProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

    try (Admin admin = Admin.create(adminProps)) {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupId);
        ConfigEntry configEntry = new ConfigEntry("share.auto.offset.reset", "earliest");

        Map<ConfigResource, Collection<AlterConfigOp>> configs = Map.of(
            configResource, List.of(new AlterConfigOp(configEntry, AlterConfigOp.OpType.SET))
        );

        admin.incrementalAlterConfigs(configs).all().get();
    }
}

记录确认

Spring Kafka 通过 ContainerProperties.ShareAckMode 暴露三种确认模式，控制记录在处理后如何被确认。

EXPLICIT（默认）：容器管理的确认

在 EXPLICIT 模式下，容器完全负责确认。监听器正常返回后，容器为该记录发送 ACCEPT。监听器抛异常时，由 ShareConsumerRecordRecoverer 决定结果（ACCEPT、RELEASE 或 REJECT；默认：REJECT）。这最接近于在常规消费者上禁用 auto.commit——容器管理所有确认决策，无需监听器参与。

这是默认模式；无需额外配置：

@Bean
public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerContainerFactory(
        ShareConsumerFactory<String, String> shareConsumerFactory) {
    // EXPLICIT is the default — no shareAckMode configuration needed
    return new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
}

@KafkaListener(topics = "my-queue-topic", containerFactory = "shareKafkaListenerContainerFactory",
               groupId = "my-share-group")
public void listen(ConsumerRecord<String, String> record) {
    // Container sends ACCEPT on success; recoverer decides on error (default: REJECT)
    process(record.value());
}

MANUAL：监听器管理的确认

在 MANUAL 模式下，监听器驱动每条记录的每个确认决策。每条记录都附带一个非空的 ShareAcknowledgment 实例投递给监听器，监听器必须用一个终止操作（acknowledge()、release() 或 reject()）恰好调用一次。后续轮询会被阻塞，直到上一次轮询的所有记录都被确认。

当你的业务逻辑逐条记录地决定确认结果时，使用 MANUAL 模式。

@Bean
public ShareKafkaListenerContainerFactory<String, String> manualShareKafkaListenerContainerFactory(
        ShareConsumerFactory<String, String> shareConsumerFactory) {
    ShareKafkaListenerContainerFactory<String, String> factory =
        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
    factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);
    return factory;
}

监听器必须接受一个 ShareAcknowledgment 参数（即实现 AcknowledgingShareConsumerAwareMessageListener）：

@KafkaListener(topics = "order-queue", containerFactory = "manualShareKafkaListenerContainerFactory")
public void processOrder(ConsumerRecord<String, String> record, ShareAcknowledgment acknowledgment) {
    try {
        processOrderLogic(record.value());
        acknowledgment.acknowledge(); // ACCEPT — success
    }
    catch (TransientException e) {
        acknowledgment.release();     // RELEASE — redelivery
    }
    catch (Exception e) {
        acknowledgment.reject();      // REJECT — permanent failure
    }
}

important

ShareAckMode.MANUAL 要求监听器接受一个 ShareAcknowledgment 参数（即实现 AcknowledgingShareConsumerAwareMessageListener）。容器在启动时校验这一点，如果使用普通 MessageListener 就抛出 IllegalStateException。

IMPLICIT：Kafka 客户端隐式模式

在 IMPLICIT 模式下，确认完全委托给 Kafka 代理，无论处理结果如何，代理都自动接受所有已获取的记录。没有可用的逐条记录确认 API；ShareAcknowledgment 参数始终为 null。这直接等价于在 Kafka 客户端配置中设置 share.acknowledgement.mode=implicit。

@Bean
public ShareKafkaListenerContainerFactory<String, String> implicitShareKafkaListenerContainerFactory(
        ShareConsumerFactory<String, String> shareConsumerFactory) {
    ShareKafkaListenerContainerFactory<String, String> factory =
        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
    factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.IMPLICIT);
    return factory;
}

注意

在 IMPLICIT 模式下，ShareConsumerRecordRecoverer 不会被调用。从代理的视角看，处理错误被静默吸收——所有记录总是被 ACCEPT。仅在不需要逐条记录投递保证时使用此模式。

确认模式对比

模式	谁来确认	成功时	监听器出错时
EXPLICIT（默认）	容器	容器发送 ACCEPT	恢复器决定（默认 REJECT）
MANUAL	监听器代码	监听器调用 acknowledge()	监听器调用 release() 或 reject()
IMPLICIT	Kafka 代理	代理自动 ACCEPT	代理自动 ACCEPT（无恢复）

从 4.0 迁移

在 4.0 中，共享消费者确认由 setExplicitShareAcknowledgment(boolean) 控制。在 4.1 中，它被 setShareAckMode(ShareAckMode) 取代。被废弃的方法仍然有效：true 映射到 MANUAL，false 映射到 EXPLICIT。

默认行为不变。 4.0 的默认值（setExplicitShareAcknowledgment(false)）是容器管理的确认，正是 ShareAckMode.EXPLICIT 所做的。使用默认值的应用无需任何改动。

如果你使用了 setExplicitShareAcknowledgment(true)，请替换它：

// Before (4.0)
factory.getContainerProperties().setExplicitShareAcknowledgment(true);

// After (4.1)
factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);

如果你在工厂配置中设置了 share.acknowledgement.mode=implicit（通过 ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG），请注意在 4.0 中此设置不生效，因为容器无论如何总是调用 consumer.acknowledge()。在 4.1 中，容器检测到此冲突，记录一条警告，并用显式模式覆盖该工厂设置。要使用 Kafka 客户端隐式模式，请在容器上显式设置：

factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.IMPLICIT);

确认类型

共享消费者支持四种确认类型：

• ACCEPT——记录处理成功；标记为已完成。
• RELEASE——临时失败；使该记录可被重新投递给本消费者或另一个消费者。
• REJECT——永久失败；归档该记录，不再投递。
• RENEW——延长获取锁（非终止）。当处理可能超过代理的锁时长（group.share.record.lock.duration.ms，默认 30 秒）时使用 renew()。仍然需要一个终止确认（acknowledge()、release() 或 reject()）。

ShareAcknowledgment API

ShareAcknowledgment 接口仅在 ShareAckMode.MANUAL 模式下非空：

public interface ShareAcknowledgment {
    void acknowledge(); // ACCEPT — record successfully processed
    void release();     // RELEASE — redelivery on transient failure
    void reject();      // REJECT — permanent failure, do not retry
    void renew();       // RENEW (non-terminal) — extend acquisition lock
}

监听器接口

基础消息监听器

在 EXPLICIT（默认）或 IMPLICIT 模式下使用标准的 MessageListener：

@KafkaListener(topics = "my-topic", containerFactory = "shareKafkaListenerContainerFactory")
public void listen(ConsumerRecord<String, String> record) {
    // In EXPLICIT mode (default): container sends ACCEPT on success, REJECT on error
    process(record.value());
}

AcknowledgingShareConsumerAwareMessageListener

此接口提供对 ShareConsumer 实例的访问，并在 MANUAL 模式下提供对 ShareAcknowledgment 的访问。acknowledgment 参数仅在 MANUAL 模式下非空；在 EXPLICIT 和 IMPLICIT 模式下为 null。

EXPLICIT 模式（默认——acknowledgment 为 null）

@KafkaListener(
    topics = "my-topic",
    containerFactory = "shareKafkaListenerContainerFactory"  // EXPLICIT by default
)
public void listen(ConsumerRecord<String, String> record,
                  @Nullable ShareAcknowledgment acknowledgment,
                  ShareConsumer<?, ?> consumer) {

    // acknowledgment is null in EXPLICIT mode — container handles ACCEPT/REJECT automatically
    process(record.value());

    // Access consumer metrics if needed
    Map<MetricName, ? extends Metric> metrics = consumer.metrics();
}

MANUAL 模式（acknowledgment 非空）

@KafkaListener(
    topics = "my-topic",
    containerFactory = "manualShareKafkaListenerContainerFactory" // ShareAckMode.MANUAL
)
public void listen(ConsumerRecord<String, String> record,
                  @Nullable ShareAcknowledgment acknowledgment,
                  ShareConsumer<?, ?> consumer) {

    // acknowledgment is non-null in MANUAL mode
    try {
        processRecord(record);
        acknowledgment.acknowledge(); // ACCEPT
    }
    catch (RetryableException e) {
        acknowledgment.release(); // Will be redelivered
    }
    catch (Exception e) {
        acknowledgment.reject(); // Permanent failure
    }
}

确认约束（MANUAL 模式）

在 ShareAckMode.MANUAL 中，容器强制以下约束：

• 轮询阻塞：后续轮询被阻塞，直到上一次轮询的所有记录都被确认。
• 终止确认：每条记录必须恰好收到一个终止确认（acknowledge()、release() 或 reject()）。在此之前你可以零次或多次调用 renew() 以延长获取锁。
• 错误处理：如果处理抛出异常，结果由 ShareConsumerRecordRecoverer 决定（默认：REJECT）。

注意

在 ShareAckMode.MANUAL 中，未确认记录会阻塞进一步的消息处理。务必确保在所有代码路径（包括异常处理器）中都确认了记录。

确认超时检测

为帮助识别缺失的确认，Spring Kafka 在 MANUAL 模式下提供可配置的超时检测。当一条记录在指定超时内未被确认时，会记录一条警告，附带该未确认记录的详细信息。

@Bean
public ShareKafkaListenerContainerFactory<String, String> manualShareKafkaListenerContainerFactory(
    ShareConsumerFactory<String, String> shareConsumerFactory) {
    ShareKafkaListenerContainerFactory<String, String> factory =
        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
    factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);

    // Set acknowledgment timeout (default is 30 seconds)
    factory.getContainerProperties().setShareAcknowledgmentTimeout(Duration.ofSeconds(30));

    return factory;
}

当一条记录超过超时时，你会看到类似这样的警告：

WARN: Record not acknowledged within timeout (30 seconds).
In ShareAckMode.MANUAL you must call ack.acknowledge(), ack.release(),
or ack.reject() for every record (call ack.renew() to extend the lock; a terminal ack is still required).
Unacknowledged record: topic='my-topic', partition=0, offset=42

此特性有助于在缺失的确认调用因轮询阻塞而导致 "没有新记录被消费" 的症状之前，将其识别出来。

同步与异步提交

默认情况下，共享容器使用 commitSync() 以确保在继续之前确认已持久化。当稍微放宽确认持久性以换取更高吞吐可以接受时，将 syncShareCommits 设为 false 以使用 commitAsync()。

@Bean
public ShareKafkaListenerContainerFactory<String, String> asyncCommitFactory(
        ShareConsumerFactory<String, String> consumerFactory) {
    ShareKafkaListenerContainerFactory<String, String> factory =
        new ShareKafkaListenerContainerFactory<>(consumerFactory);
    factory.getContainerProperties().setSyncShareCommits(false); // use commitAsync()
    return factory;
}

属性值	调用的 Kafka API	权衡
true（默认）	consumer.commitSync()	持久——保证确认在下一次轮询之前已持久化
false	consumer.commitAsync()	吞吐更高——如果消费者在异步提交完成前崩溃，则有轻微的重新投递风险

错误处理

共享容器在两个层面处理错误：poll() 期间（轮询级）和监听器抛出异常时（监听器级）。

轮询级错误处理

如果 consumer.poll() 抛出异常，容器会处理它，使消费者线程不停止：

• RecordDeserializationException：当一条记录无法反序列化时抛出（例如 StringDeserializer 遇到无效 UTF-8）。容器捕获它，记录一条警告，并继续轮询。在 EXPLICIT 和 MANUAL 模式下，容器使用异常中的主题、分区和偏移量 REJECT 受影响的记录。在 IMPLICIT 模式下不进行显式确认调用——代理自动接受该记录。后续记录正常处理。
• CorruptRecordException：当 CRC 校验失败时抛出（例如启用了 check.crcs）。Kafka 客户端自动拒绝该损坏批。容器捕获该异常，记录它，并继续轮询。

如果没有这种处理，单条坏记录就会终止消费者线程。

监听器级错误处理：ShareConsumerRecordRecoverer

当监听器抛出异常时，容器委托给一个 ShareConsumerRecordRecoverer 来决定是 ACCEPT、RELEASE 还是 REJECT 失败的记录。默认是 ShareConsumerRecordRecoverer.REJECTING（记录日志并 REJECT）；也提供 ShareConsumerRecordRecoverer.RELEASING（记录日志并 RELEASE 以重新投递）。

你可以提供一个自定义恢复器，对临时失败（例如下游超时）RELEASE 记录，以便它们能被重新投递给另一个消费者，同时 REJECT 永久失败。

配置自定义恢复器

在容器或工厂上设置恢复器，使其适用于该工厂创建的所有容器：

@Bean
public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerContainerFactory(
        ShareConsumerFactory<String, String> shareConsumerFactory) {

    ShareKafkaListenerContainerFactory<String, String> factory =
        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);

    factory.setShareConsumerRecordRecoverer((record, ex) -> {
        if (ex instanceof TransientException || ex.getCause() instanceof TimeoutException) {
            return AcknowledgeType.RELEASE;
        }
        return AcknowledgeType.REJECT;
    });

    return factory;
}

你也可以通过 container.setShareConsumerRecordRecoverer(recoverer) 在单个容器上设置恢复器。

恢复器行为

• 恢复器不得返回 RENEW；容器把它视为 REJECT。
• 如果恢复器本身抛出异常，容器对该记录回退到 REJECT。
• 默认恢复器以 ERROR 级别记录日志，并对每个异常返回 REJECT。

与毒消息保护的关系

代理端投递计数（见 毒消息保护与投递计数）限制了一条记录可被获取的次数。即使你的恢复器总是返回 RELEASE，代理最终也会在达到配置的限制（默认 5）后归档该记录，因此毒消息不会无限循环。

与传统容器错误处理的区别

共享消费者不使用 CommonErrorHandler。该接口是为基于分区的消费者设计的（seek、剩余记录、偏移量提交）。共享消费者使用基于确认的恢复：恢复器只决定 ACCEPT、RELEASE 或 REJECT；容器执行确认。共享容器中没有 seek 或死信主题集成。

确认示例（MANUAL 模式）

混合确认模式

@KafkaListener(topics = "order-processing", containerFactory = "manualShareKafkaListenerContainerFactory")
public void processOrder(ConsumerRecord<String, String> record, ShareAcknowledgment acknowledgment) {
    String orderData = record.value();
    try {
        if (isValidOrder(orderData)) {
            if (processOrder(orderData)) {
                acknowledgment.acknowledge(); // ACCEPT — success
            }
            else {
                acknowledgment.release(); // RELEASE — temporary failure, retry later
            }
        }
        else {
            acknowledgment.reject(); // REJECT — invalid order, don't retry
        }
    }
    catch (Exception e) {
        acknowledgment.reject(); // REJECT — unexpected failure
    }
}

条件确认

@KafkaListener(topics = "data-validation", containerFactory = "manualShareKafkaListenerContainerFactory")
public void validateData(ConsumerRecord<String, String> record, ShareAcknowledgment acknowledgment) {
    ValidationResult result = validator.validate(record.value());
    switch (result.getStatus()) {
        case VALID -> acknowledgment.acknowledge();        // ACCEPT
        case INVALID_RETRYABLE -> acknowledgment.release(); // RELEASE
        case INVALID_PERMANENT -> acknowledgment.reject();  // REJECT
    }
}

毒消息保护与投递计数

KIP-932 包含代理端毒消息保护，以防止无法处理的记录被无休止地重新投递。

工作原理

每当一条记录被共享组中的一个消费者获取时，代理就递增一个内部投递计数。第一次获取把投递计数设为 1，之后每次获取递增它。当投递计数达到配置的限制（默认：5）时，该记录转为 Archived（已归档）状态，不再具备额外投递尝试的资格。

important

投递计数不暴露给应用 投递计数由代理内部维护，不暴露给消费者应用。这是 KIP-932 中有意为之的设计决策。投递计数是近似值，作为毒消息保护机制，而非精确的重新投递计数器。应用无法通过任何 API 查询或访问此值。

对于应用级重试逻辑，使用确认类型：

• RELEASE——使记录可被重新投递（计入投递计数）
• REJECT——标记为永久失败（不引起重新投递）
• ACCEPT——成功处理（不引起重新投递）

一旦达到 group.share.delivery.count.limit，代理自动防止无休止的重新投递，把该记录转为 Archived 状态。

重试策略建议

以下是一个示例，说明如何根据异常类型使用各种确认类型。

@KafkaListener(topics = "orders", containerFactory = "manualShareKafkaListenerContainerFactory")
public void processOrder(ConsumerRecord<String, String> record, ShareAcknowledgment ack) {
    try {
        // Attempt to process the order
        orderService.process(record.value());
        ack.acknowledge(); // ACCEPT - successfully processed
    }
    catch (TransientException e) {
        // Temporary failure (network issue, service unavailable, etc.)
        // Release the record for redelivery
        // Broker will retry up to group.share.delivery.count.limit times
        logger.warn("Transient error processing order, will retry: {}", e.getMessage());
        ack.release(); // RELEASE - make available for retry
    }
    catch (ValidationException e) {
        // Permanent semantic error (invalid data format, business rule violation, etc.)
        // Do not retry - this record will never succeed
        logger.error("Invalid order data, rejecting: {}", e.getMessage());
        ack.reject(); // REJECT - permanent failure, do not retry
    }
    catch (Exception e) {
        // Unknown error - typically safer to reject to avoid infinite loops
        // But could also release if you suspect it might be transient
        logger.error("Unexpected error processing order, rejecting: {}", e.getMessage());
        ack.reject(); // REJECT - avoid poison message loops
    }
}

代理的毒消息保护确保，即使你对错误总是使用 RELEASE，记录也不会被无休止地重试。它们在达到投递尝试限制后会自动归档。

与常规消费者的区别

共享消费者在几个关键方面与常规消费者不同：

1. 无分区分配：共享消费者不能被分配特定分区
2. 无主题模式：共享消费者不支持订阅主题模式
3. 协作消费：同一共享组中的多个消费者可以同时从相同分区消费
4. 记录级确认：支持带 ACCEPT、RELEASE 和 REJECT 类型的显式确认
5. 不同的组管理：共享组使用不同的协调器协议
6. 无批处理：共享消费者逐条处理记录，而非成批处理
7. 代理端重试管理：投递计数跟踪和毒消息保护由代理管理，不暴露给应用

限制与注意事项

当前限制

• 无消息转换器：共享消费者尚不支持消息转换器
• 无批量监听器：共享消费者不支持批量处理
• 轮询约束：在 ShareAckMode.MANUAL 中，每个消费者线程内未确认的记录会阻塞后续轮询