这之所以是个问题,是由于 consumer 的非线程安全的,他不被允许在多个线程中使用,所以直接在 shutdownHook线程 中调用 consumer 的 close()
方法,会抛出相关异常。
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1824)
at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:1713)
at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:1692)
如果你没有在程序退出前很好的 close consumer。最明显的行为就是在下次启动程序消费数据的时候会发现为consumer分配分区的过程可能非常慢(几分钟),还有一个问题是Kafka不能立刻知道 consumer 已经退出,如果同组在其他地方有其他消费者,再均衡在消费心跳超时后才会触发。其他可能的问题还不知道,水平有限。总之,close() 是必须的。
解决思路
解决切入点,是使用 consumer 的 wakeup()
来完成,这个方法是可以在多线程下调用的。
主要思路是在shutdownHook线程中使用 consumer 的 wakeup()
方法让主线程中的消费者在下次 poll 的时候抛出 WakeupException,以退出 while 循环,在 finally 中关闭 consumer。当然是还需要在shutdownHook线程中等待主线程完成任务再退出整个程序。
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
System.out.println("Starting exit...");
consumer.wakeup();
try {
// 主线程继续执行,以便可以关闭consumer,提交偏移量
mainThread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record : records) {
System.out.println("topic = " + record.topic() + ", partition = " + record.partition()
+ ", offset = " + record.offset());
}
consumer.commitAsync();
}
}catch (WakeupException e) {
// 不处理异常
} finally {
// 在退出线程之前调用consumer.close()
consumer.close();
System.out.println("Closed consumer and we are done");
}
参考 https://www.jianshu.com/p/6e89db47c116
版权声明:本文为u013054888原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接和本声明。