多线程消费 KafkaConsumer is not safe for multi-threaded access

√锋²º¹8ヾ/❤ 发表于: 2018-09-05 最后更新时间: 2018-09-05 09:05:17 12,319 游览

一个副本、一个分区生产信息。

消费的时候，开启了3个线程，并发消费同一主题，同一个组。

启动后瞬间就出现：

Exception in thread "Thread-10" java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1674)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1031)
at com.oristartech.kafka.core.consumer.ConsumerHandler$1.run(ConsumerHandler.java:100)
at java.lang.Thread.run(Thread.java:748)

翻看了一下其他问题发问，好像是说消费者是非线程安全的，而且线程和分区也有着限定的关系（线程 ≤ 分区）。

如果要满足多线程消费同主题+同分组，该如何处理才好

kafka

发表于 2018-09-05

√锋²º¹8ヾ/❤

添加评论

一个消费者对应一个分区，所以你多个消费者，其实永远只有一个消费者拿到消息.
kafka拉取消息是一批一批的，大概一次拉取2000-4000条。你应该在拉取后进行多线程处理.

回答于 6年前

半兽人

√锋²º¹8ヾ/❤ -> 半兽人 6年前

好的。
但是，poll的时候，测试发现大概每轮只有500条，就停顿0.5-1秒。并不像您说的“大概一次拉取2000-4000条”。
消费者拉取主函数：
public void consume(int threadNumber) {

  if (ConsumerThreadTools.mapListeners.isEmpty()) {
   logger.error("消息处理对象为空，不进行消费.........返回");
   return;
  }

  if(executors == null) {

   executors = new ThreadPoolExecutor(threadNumber, threadNumber, 0L, TimeUnit.MILLISECONDS,
     new ArrayBlockingQueue<Runnable>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
  }else {

  }

  try {
   while (true) {
    ConsumerRecords<String, String> records = consumer.poll(500);
    System.out.println(JSONObject.toJSONString(records));
    if (!records.isEmpty()) {
     executors.submit(new ConsumerThreadWorker(records, offsets));
    }
    commitOffsets();
   }
  } catch (WakeupException e) {
   // swallow this exception
  } finally {
   commitOffsets();
   consumer.close();
  }
}

半兽人 -> √锋²º¹8ヾ/❤ 6年前

拉取的条数跟你消息大小有关，你把poll时间调成100，提交改成自动试试

你的答案

查看kafka相关的其他问题或提一个您自己的问题。

提问

找不到想要的答案？提一个您自己的问题。

多线程消费 KafkaConsumer is not safe for multi-threaded access

你的答案

昵称