java多线程消费kafka

在Java中消费Kafka可以使用Kafka Consumer API。如果需要使用多线程消费Kafka，可以采用以下步骤：

1. 创建一个Kafka Consumer实例，指定消费者组和Kafka Broker地址。

2. 创建多个线程，每个线程都创建一个Kafka Consumer实例，并订阅相同的Topic。同时，每个线程指定不同的Consumer Group ID。

3. 在每个线程中循环消费Kafka消息。可以使用轮询方式或者阻塞方式来消费消息。

4. 在消费消息的过程中，需要注意线程安全问题。可以使用线程安全的数据结构来保存消费得到的消息。

以下是一个示例代码：

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

import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerThread implements Runnable {
    private final KafkaConsumer<String, String> consumer;
    private final String topic;

    public KafkaConsumerThread(String brokers, String groupId, String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", brokers);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        this.consumer = new KafkaConsumer<>(props);
        this.topic = topic;
    }

    public void run() {
        consumer.subscribe(Arrays.asList(this.topic));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                // TODO: 处理消费消息
            }
        }
    }
}

public class KafkaMultiThreadConsumer {
    private static final int NUM_THREADS = 3;
    private static final String TOPIC_NAME = "test-topic";
    private static final String GROUP_ID = "test-group";
    private static final String BROKERS = "localhost:9092";

    public static void main(String[] args) {
        for (int i = 0; i < NUM_THREADS; i++) {
            KafkaConsumerThread consumerThread = new KafkaConsumerThread(BROKERS, GROUP_ID + i, TOPIC_NAME);
            new Thread(consumerThread).start();
        }
    }
}

在上面的示例代码中，KafkaConsumerThread是一个消费Kafka消息的线程类。在main函数中，创建了多个KafkaConsumerThread实例，并且每个实例都使用不同的Consumer Group ID。这样，就可以实现多线程消费Kafka。