Kafka集群中的多个消费者

在Kafka集群中，可以有多个消费者同时使用同一个消费者组来消费消息。以下是一个使用Java代码示例的解决方法：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class MultiConsumerExample {
    private static final String TOPIC_NAME = "your_topic_name";
    private static final String GROUP_ID = "your_group_id";
    private static final String BOOTSTRAP_SERVERS = "your_bootstrap_servers";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC_NAME));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("Received message: " + record.value());
                    // 进行消费逻辑处理
                }
                consumer.commitAsync(); // 异步提交消费位移
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
}

在上面的示例中，我们创建了一个KafkaConsumer实例，并设置了一些必要的属性，如BOOTSTRAP_SERVERS_CONFIG（Kafka集群的地址）、GROUP_ID_CONFIG（消费者组的ID）等。然后，我们通过subscribe方法订阅了一个主题。接下来，在一个循环中，我们使用poll方法从Kafka集群中获取消息，然后进行消费逻辑处理。最后，我们使用commitAsync方法异步提交消费位移，以确保消息被正确消费。

你可以根据自己的需求，对示例代码进行修改和扩展。