java - Java Nio 上的 Kafka 消费者心跳线程块

Question

使用的软件库版本：Spring 2.2.3.Release、Spring kafka 2.x、Kafka-Client 2.0.0 Java 版本：OpenJdk 1.8 平台 Kafka 版本：Apache kafka 2.2.0 ( https://docs.confluent.io /5.2.1/release-notes.html#apache-kafka-2-2-0-cp2 )

配置：主题有 6 个分区。数据速率：传入数据速率为 3-4 msgs/秒。

场景：使用上述配置，测试连续运行 3 天。2.5天后，我在我们的消费者组中遇到了没有收到来自主题的消息的问题。经过详细调查，我发现消费者线程被阻塞了，正好有18个线程处于阻塞状态。此外，线程图表明消费者线程正在等待“Java.nio”。

PFA 日志和屏幕截图。a.) 日志 b.) 截图

a) 日志

"kafka-coordinator-heartbeat-thread | release-registry-group" #128 daemon prio=5 os_prio=0 tid=0x00007f1954001800 nid=0x8a waiting for monitor entry [0x00007f197f7f6000]
   java.lang.Thread.State: BLOCKED (on object monitor)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$HeartbeatThread.run(AbstractCoordinator.java:1005)
        - waiting to lock <0x00000000c3382070> (a org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)

"org.springframework.kafka.KafkaListenerEndpointContainer#1-2-C-1" #127 prio=5 os_prio=0 tid=0x00007f1b6dbc1800 nid=0x89 runnable [0x00007f197f8f6000]
   java.lang.Thread.State: RUNNABLE
        at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
        at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
        at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
        at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
        - locked <0x00000000c33822d8> (a sun.nio.ch.Util$3)
        - locked <0x00000000c33822c8> (a java.util.Collections$UnmodifiableSet)
        - locked <0x00000000c33822e8> (a sun.nio.ch.EPollSelectorImpl)
        at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
        at org.apache.kafka.common.network.Selector.select(Selector.java:689)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:409)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:510)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:271)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:242)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:218)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:230)
        - locked <0x00000000c3382070> (a org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:314)
        at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1218)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1175)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1154)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:732)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:689)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.lang.Thread.run(Thread.java:748)

b.) 截图

我想了解根本原因和可能的解决方案来解决它，并提前致谢。

标签： javamultithreadingapache-kafkaspring-kafkaconfluent-platform