查看kafka消息被谁消费

在Kafka中，我们可以通过Consumer Group来查看消费者消费的消息。在同一个Consumer Group中，每个消费者都会消费同一个分区的消息。

首先，我们需要找到当前正在消费消息的消费者组。可以通过以下命令来查找：

bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

这个命令会列出所有当前正在运行的消费者组。

接下来，我们可以通过以下命令来查看某个消费者组中的消费者消费的消息：

bin/kafka-consumer-groups.sh --describe --group my-group --bootstrap-server localhost:9092

其中，my-group是需要查看的消费者组的名称。

这个命令将会列出消费者组中每个消费者消费的分区信息，包括分区的名称、消费者的ID、消费的偏移量等信息。

如果需要查看某个特定分区的消费情况，可以运行以下命令：

bin/kafka-consumer-groups.sh --describe --group my-group --topic my-topic --bootstrap-server localhost:9092

其中，my-topic是需要查看的分区名称。

除了通过命令行的方式来查看消费者消费的消息外，我们还可以通过Java代码来实现。以下是一个简单的示例代码：

import java.util.*;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.*;
import org.apache.kafka.common.serialization.*;

public class KafkaConsumerExample {
    private static final String TOPIC_NAME = "my-topic";
    private static final String GROUP_ID = "my-group";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", GROUP_ID);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(TOPIC_NAME));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Topic: %s, Partition: %d, Offset: %d, Key: %s, Value: %s\n",
                            record.topic(),