Spring Boot集成Parallel Consumer：ParallelStreamProcessor适配问题求助

我正在测试parallel-consumer库的运行表现，但在配置工厂容器定制器时遇到问题。我希望设置类型为ParallelStreamProcessor的消息监听器，然而setupMessageListener方法仅接受AcknowledgingMessageListener或MessageListener类型。

我知道可以通过继承现有监听器创建自定义监听器来解决，但希望获得其他替代方案。

相关方法定义：

/**
 * Set the message listener; must be a {@link org.springframework.kafka.listener.MessageListener}
 * or {@link org.springframework.kafka.listener.AcknowledgingMessageListener}.
 * @param messageListener the listener.
 */
public void setMessageListener(Object messageListener) {
    this.messageListener = messageListener;
    adviseListenerIfNeeded();
}

我的Bean创建代码：

@Bean
public ParallelStreamProcessor<String, String> parallelStreamProcessor(Consumer<String, String> kafkaConsumer) {

    ParallelConsumerOptions<String, String> options = ParallelConsumerOptions.<String, String>builder()
        .ordering(ParallelConsumerOptions.ProcessingOrder.KEY)
        .maxConcurrency(16)
        .commitMode(ParallelConsumerOptions.CommitMode.PERIODIC_CONSUMER_SYNC)
        .consumer(kafkaConsumer)
        .build();

    return ParallelStreamProcessor.createEosStreamProcessor(options);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
    ConsumerFactory<String, String> consumerFactory,
    ParallelStreamProcessor<String, String> parallelStreamProcessor) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);

    factory.setContainerCustomizer(container -> {
      container.setupMessageListener(parallelStreamProcessor);
      container.setAutoStartup(true);
    });

    return factory;
}

替代方案 #

方案1：自定义适配器包装ParallelStreamProcessor #

创建实现AcknowledgingMessageListener的适配器类，将消息转发给ParallelStreamProcessor，兼容Spring Kafka API的同时避免继承现有监听器：

public class ParallelStreamProcessorAdapter<K, V> implements AcknowledgingMessageListener<K, V> {

    private final ParallelStreamProcessor<K, V> processor;

    public ParallelStreamProcessorAdapter(ParallelStreamProcessor<K, V> processor) {
        this.processor = processor;
    }

    @Override
    public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment) {
        processor.poll(record);
        // 若使用手动提交模式，可在此调用acknowledgment.acknowledge()；PERIODIC模式下无需手动处理
    }
}

修改容器定制器代码：

factory.setContainerCustomizer(container -> {
    AcknowledgingMessageListener<String, String> adapter = new ParallelStreamProcessorAdapter<>(parallelStreamProcessor);
    container.setupMessageListener(adapter);
    container.setAutoStartup(true);
});

方案2：使用ParallelConsumer独立运行模式 #

跳过Spring Kafka容器管理，让ParallelConsumer自行启动运行，完全规避API兼容性问题：

@Bean(initMethod = "start", destroyMethod = "close")
public ParallelStreamProcessor<String, String> parallelStreamProcessor(Consumer<String, String> kafkaConsumer) {
    ParallelConsumerOptions<String, String> options = ParallelConsumerOptions.<String, String>builder()
        .ordering(ParallelConsumerOptions.ProcessingOrder.KEY)
        .maxConcurrency(16)
        .commitMode(ParallelConsumerOptions.CommitMode.PERIODIC_CONSUMER_SYNC)
        .consumer(kafkaConsumer)
        .topics(Collections.singletonList("target-topic")) // 订阅目标主题
        .build();

    ParallelStreamProcessor<String, String> processor = ParallelStreamProcessor.createEosStreamProcessor(options);
    
    // 定义消息处理逻辑
    processor.stream().forEach(record -> {
        // 编写你的业务处理代码
        System.out.println("Processing record: " + record.value());
    });

    return processor;
}

此方式无需配置ConcurrentKafkaListenerContainerFactory，ParallelConsumer会自行管理消费线程、提交逻辑等。

方案3：利用Spring Kafka原生MessageListenerAdapter #

借助Spring Kafka提供的MessageListenerAdapter，直接适配ParallelStreamProcessor的方法作为监听器入口：

factory.setContainerCustomizer(container -> {
    // 适配ParallelStreamProcessor的poll方法作为消息处理入口
    MessageListenerAdapter adapter = new MessageListenerAdapter(parallelStreamProcessor, "poll");
    container.setupMessageListener(adapter);
    container.setAutoStartup(true);
});

注意需确保ParallelStreamProcessor的poll方法参数与ConsumerRecord类型匹配，此方式适配速度快，但灵活性不如自定义适配器。

内容的提问来源于stack exchange，提问作者Amber Pandey