117笔记问答Kafka的异步回调可以通过使用Kafka消费者（Consumer）的poll()方法来实现。这个方法会返回一个ConsumerRecords对象，其中包含了从Kafka分区的最新记录。你可以为每个分区创建一个线程来处理消息，从而实现异步处理。
以下是一个简单的示例，展示了如何使用Kafka消费者实现异步回调：
- 首先，确保你已经添加了Kafka客户端依赖。如果你使用的是Maven，可以在
pom.xml文件中添加以下依赖：
org.apache.kafka kafka-clients 2.8.0
- 创建一个Kafka消费者实例，并配置相关参数，如Bootstrap服务器、组ID等：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
- 订阅感兴趣的主题：
consumer.subscribe(Arrays.asList("my-topic"));
- 创建一个线程池来处理消息：
ExecutorService executorService = Executors.newFixedThreadPool(10);
- 使用
poll()方法异步获取消息，并在单独的线程中处理它们：
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord record : records) {
executorService.submit(() -> {
// 在这里处理消息
System.out.printf("Offset = %d, Key = %s, Value = https://www.yisu.com/ask/%s%n", record.offset(), record.key(), record.value());
});
}
}
- 最后，不要忘记在程序结束时关闭消费者和线程池：
consumer.close(); executorService.shutdown();
这个示例展示了如何使用Kafka消费者实现异步回调。你可以根据自己的需求修改这个示例，例如使用不同的序列化器、处理逻辑等。