当并发=分区计数时，spring集成kafka侦听器线程读取多个分区

Question

我设置了一个Spring集成流程来处理一个有3个分区的主题，并将侦听器容器的并发性设置为3。正如所料，我看到三个线程处理来自所有3个分区的批处理。然而，我发现在某些情况下，一个侦听器线程可能处理包含来自多个分区的消息的单个批处理。在kafka中，我的数据是按id划分的，因此它可以与其他id同时处理，但不能在另一个线程上与相同的id一起处理(我很惊讶地发现这种情况正在发生)。通过阅读文档，我认为每个线程都将被分配一个分区。我正在使用这样的KafkaMessageDrivenChannelAdapter:

private static final Class<List<MyEvent>> payloadClass = (Class<List<MyEvent>>)(Class) List.class;

public KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<String, MyEvent> myChannelAdapterSpec() {
        return Kafka.messageDrivenChannelAdapter(tstatEventConsumerFactory(),
                KafkaMessageDrivenChannelAdapter.ListenerMode.batch, "my-topic") //3 partitions
                .configureListenerContainer(c -> {
                    c.ackMode(ContainerProperties.AckMode.BATCH);
                    c.id(_ID);
                    c.concurrency(3);
                    RecoveringBatchErrorHandler errorHandler = new RecoveringBatchErrorHandler(
                            (record, exception) -> log.error("failed to handle record at offset {}: {}",
                                    record.offset(), record.value(), exception),
                            new FixedBackOff(FixedBackOff.DEFAULT_INTERVAL, 2)
                    );
                    c.errorHandler(errorHandler);
                });
    }
@Bean
public IntegrationFlow myIntegrationFlow() {
        return IntegrationFlows.from(myChannelAdapterSpec())
                .handle(payloadClass, (payload, headers) -> {
                    service.performSink(payload);
                    return null;
                })
                .get();
    }

如何设置它，使每个侦听器容器线程只处理来自一个分区的消息？

Answer 1

但是，是否还有一种方法可以防止从多个分区获取带有消息的批处理，即使确实发生了重新平衡？

消费者团体不是这么运作的。如果你想拥有一个“粘性”消费者，那么考虑使用手动分配。请参见基于< code>TopicPartitionOffset的通道适配器工厂...主题分区:

/**
 * Create an initial
 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
 * @param consumerFactory the {@link ConsumerFactory}.
 * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}.
 * @param topicPartitions the {@link TopicPartitionOffset} vararg.
 * @param <K> the Kafka message key type.
 * @param <V> the Kafka message value type.
 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
 */
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
        ConsumerFactory<K, V> consumerFactory,
        KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode,
        TopicPartitionOffset... topicPartitions) {

然后，它不会被视为使用者组，您必须创建多个通道适配器，每个适配器都指向其特定分区。所有这些通道适配器都可以向同一消息通道发出消息。