Kafka消费者消息处理

1. 创建一个Kafka消费者，使用poll()方法从订阅的主题中获取消息。
2. 读取每个消息并处理该消息。
3. 在处理消息时，可以进行各种操作，如存储数据、转发消息，或者执行其他自定义的逻辑。
4. 如果处理过程中出现异常，应该捕获并处理该异常，以确保消费者能够继续消费其他消息。
5. 处理完成后，通过调用commitSync()或commitAsync()方法来提交偏移量。
6. 如果需要在消费者端进行多线程处理，可以使用并发API中提供的并发工具（如ConcurrentHashMap、AtomicInteger等）来实现多线程处理。

示例代码：

import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.KafkaConsumer; import java.util.Arrays; import java.util.Properties;

public class ConsumerExample { public static void main(String[] args) { String topicName = "test"; Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test-group"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); Consumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList(topicName)); while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { System.out.println("Received message: " + record.value()); //处理该消息 } consumer.commitAsync();//异步提交偏移量 } } }