spring整合kafka 消费者 异步处理方式

在Spring整合Kafka中，消费者可以使用异步处理方式处理消息。具体实现可以使用Spring提供的@KafkaListener注解，通过设置concurrency属性来指定消费者线程池的大小，从而实现异步处理。示例代码如下：

@KafkaListener(topics = "${kafka.topic}", containerFactory = "kafkaListenerContainerFactory", concurrency = "5")
public void onMessage(ConsumerRecord<String, String> record) {
    // 异步处理消息
    CompletableFuture.runAsync(() -> {
        // 处理消息逻辑
    });
}

在上述代码中，我们将@KafkaListener注解的concurrency属性设置为5，表示使用5个线程并发消费消息。在onMessage方法中，我们使用CompletableFuture.runAsync方法将消息处理逻辑异步执行，从而不会阻塞消费者线程，提升消费者的处理能力。