如何在kafka 0.9.0中使用多线程消费者？ [英] How to use multi-thread consumer in kafka 0.9.0?

问题描述

kafka的文档给出了以下描述的方法：

The doc of kafka give an approach about with following describes:

每个线程一个消费者：一个简单的选择是给每个线程它自己的消费者>实例。

One Consumer Per Thread:A simple option is to give each thread its own consumer > instance.

我的代码：

public class KafkaConsumerRunner implements Runnable {

    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final CloudKafkaConsumer consumer;
    private final String topicName;

    public KafkaConsumerRunner(CloudKafkaConsumer consumer, String topicName) {
        this.consumer = consumer;
        this.topicName = topicName;
    }

    @Override
    public void run() {
        try {
            this.consumer.subscribe(topicName);
            ConsumerRecords<String, String> records;
            while (!closed.get()) {
                synchronized (consumer) {
                    records = consumer.poll(100);
                }
                for (ConsumerRecord<String, String> tmp : records) {
                    System.out.println(tmp.value());
                }
            }
        } catch (WakeupException e) {
            // Ignore exception if closing
            System.out.println(e);
            //if (!closed.get()) throw e;
        }
    }

    // Shutdown hook which can be called from a separate thread
    public void shutdown() {
        closed.set(true);
        consumer.wakeup();
    }

    public static void main(String[] args) {
        CloudKafkaConsumer kafkaConsumer = KafkaConsumerBuilder.builder()
                .withBootstrapServers("172.31.1.159:9092")
                .withGroupId("test")
                .build();
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        executorService.execute(new KafkaConsumerRunner(kafkaConsumer, "log"));
        executorService.execute(new KafkaConsumerRunner(kafkaConsumer, "log.info"));
        executorService.shutdown();
    }
}

但它不起作用并抛出异常：

but it doesn't work and throws an exception:

java.util.ConcurrentModificationException：KafkaConsumer对多线程访问不安全

java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access

此外，我还阅读了Flink（一个用于分布式流和批量数据处理的开源平台）的来源。使用多线程消费者的Flink与我的类似。

Furthermore, I read the source of Flink (an open source platform for distributed stream and batch data processing). Flink using multi-thread consumer is similar to mine.

long pollTimeout = Long.parseLong(flinkKafkaConsumer.properties.getProperty(KEY_POLL_TIMEOUT, Long.toString(DEFAULT_POLL_TIMEOUT)));
pollLoop: while (running) {
    ConsumerRecords<byte[], byte[]> records;
    //noinspection SynchronizeOnNonFinalField
    synchronized (flinkKafkaConsumer.consumer) {
        try {
            records = flinkKafkaConsumer.consumer.poll(pollTimeout);
        } catch (WakeupException we) {
            if (running) {
                throw we;
            }
            // leave loop
            continue;
        }
    }

flli-thread of mutli-thread

出了什么问题？

推荐答案

Kafka消费者不是线程安全的。正如您在问题中指出的那样，该文件声明

Kafka consumer is not thread safe. As you pointed out in your question, the document stated that

一个简单的选择是为每个线程提供自己的消费者实例

A simple option is to give each thread its own consumer instance

但是在您的代码中，您拥有由不同KafkaConsumerRunner实例包装的相同的使用者实例。因此，多个线程正在访问同一个消费者实例。 kafka文档明确声明

But in your code, you have the same consumer instance wrapped by different KafkaConsumerRunner instances. Thus multiple threads are accessing the same consumer instance. The kafka documentation clearly stated

Kafka消费者不是线程安全的。所有网络I / O都发生在进行调用的应用程序的
线程中。用户需要
来确保多线程访问正确
同步。未同步访问将导致
ConcurrentModificationException。

The Kafka consumer is NOT thread-safe. All network I/O happens in the thread of the application making the call. It is the responsibility of the user to ensure that multi-threaded access is properly synchronized. Un-synchronized access will result in ConcurrentModificationException.

这正是您收到的异常。

That's exactly the exception you received.

这篇关于如何在kafka 0.9.0中使用多线程消费者？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！