akka流不消耗，只记录唤醒时间

vybvopom  于 2021-06-06  发布在  Kafka
关注(0)|答案(0)|浏览(225)

我有这个应用程序，我们已经测试了几个月了。通常的设置要么使用嵌入式Kafka，要么使用docker compose中的Kafka。我得承认只有1个分区。一旦我们部署了它（dockerized，连接到一个真正的代理，只有一个），它甚至不会开始消费，而是只记录以下内容：

May  2 10:40:05 ip-10-189-162-144 docker: [WARN] [05/02/2018 10:40:05.533] [KafkaIngester-akka.actor.default-dispatcher-4] [akka://KafkaIngester/system/kafka-consumer-1] KafkaConsumer poll is taking significantly longer (13000ms) to return from poll then the configured poll interval (500ms). Waking up consumer to avoid thread starvation.
May  2 10:40:05 ip-10-189-162-144 docker: [WARN] [05/02/2018 10:40:05.540] [KafkaIngester-akka.actor.default-dispatcher-4] [akka://KafkaIngester/system/kafka-consumer-1] Wake up has been triggered. Dumping stacks: Thread[server-scheduler-1,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.sleep(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: akka.actor.LightArrayRevolverScheduler.waitNanos(LightArrayRevolverScheduler.scala:85)
May  2 10:40:05 ip-10-189-162-144 docker: akka.actor.LightArrayRevolverScheduler$$anon$4.nextTick(LightArrayRevolverScheduler.scala:265)
May  2 10:40:05 ip-10-189-162-144 docker: akka.actor.LightArrayRevolverScheduler$$anon$4.run(LightArrayRevolverScheduler.scala:235)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[Finalizer,8,system]
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Object.wait(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:143)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:164)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:212)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.actor.default-dispatcher-5,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.forkjoin.ForkJoinPool.scan(ForkJoinPool.java:2075)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.kafka.default-dispatcher-111,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:467)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.kafka.default-dispatcher-116,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:467)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[persistence.database housekeeper,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.actor.default-dispatcher-4,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.dumpThreads(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.getAllStackTraces(Thread.java:1610)
May  2 10:40:05 ip-10-189-162-144 docker: akka.kafka.KafkaConsumerActor.$anonfun$poll$1(KafkaConsumerActor.scala:283)
May  2 10:40:05 ip-10-189-162-144 docker: akka.kafka.KafkaConsumerActor$$Lambda$675/1405931784.apply$mcV$sp(Unknown Source)
May  2 10:40:05 ip-10-189-162-144 docker: akka.actor.Scheduler$$anon$4.run(Scheduler.scala:140)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:43)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.kafka.default-dispatcher-117,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:467)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.kafka.default-dispatcher-113,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:467)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[server-akka.io.pinned-dispatcher-10,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
May  2 10:40:05 ip-10-189-162-144 docker: akka.io.SelectionHandler$ChannelRegistryImpl$$anon$3.tryRun(SelectionHandler.scala:128)
May  2 10:40:05 ip-10-189-162-144 docker: akka.io.SelectionHandler$ChannelRegistryImpl$Task.run(SelectionHandler.scala:246)
May  2 10:40:05 ip-10-189-162-144 docker: akka.io.SelectionHandler$ChannelRegistryImpl$$anon$3.run(SelectionHandler.scala:161)
May  2 10:40:05 ip-10-189-162-144 docker: akka.util.SerializedSuspendableExecutionContext.run$1(SerializedSuspendableExecutionContext.scala:67)
May  2 10:40:05 ip-10-189-162-144 docker: akka.util.SerializedSuspendableExecutionContext.run(SerializedSuspendableExecutionContext.scala:71)
May  2 10:40:05 ip-10-189-162-144 docker: akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[kafka-producer-network-thread | producer-1,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
May  2 10:40:05 ip-10-189-162-144 docker: sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
May  2 10:40:05 ip-10-189-162-144 docker: org.apache.kafka.common.network.Selector.select(Selector.java:672)
May  2 10:40:05 ip-10-189-162-144 docker: org.apache.kafka.common.network.Selector.poll(Selector.java:396)
May  2 10:40:05 ip-10-189-162-144 docker: org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:460)
May  2 10:40:05 ip-10-189-162-144 docker: org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:239)
May  2 10:40:05 ip-10-189-162-144 docker: org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:163)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[DestroyJavaVM,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: Thread[Abandoned connection cleanup thread,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Object.wait(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:143)
May  2 10:40:05 ip-10-189-162-144 docker: com.mysql.cj.jdbc.AbandonedConnectionCleanupThread.run(AbandonedConnectionCleanupThread.java:43)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.kafka.default-dispatcher-114,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:467)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.kafka.default-dispatcher-118,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:467)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.kafka.default-dispatcher-115,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:467)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1073)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
May  2 10:40:05 ip-10-189-162-144 docker: java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
May  2 10:40:05 ip-10-189-162-144 docker: java.lang.Thread.run(Thread.java:748)
May  2 10:40:05 ip-10-189-162-144 docker: Thread[KafkaIngester-akka.kafka.default-dispatcher-112,5,main]
May  2 10:40:05 ip-10-189-162-144 docker: sun.misc.Unsafe.park(Native Method)

.....

正如您所看到的，问题是唤醒超时。我试着增加到13秒，但还是失败了。我试图增加投票间隔，但还是失败了。我们尝试从同一台机器上的docker容器运行telnet kafka 9092，连接被接受。但它什么也不消耗。我们真的不知道为什么。
我的流程是
我的流程是：

val consumerSettings = ConsumerSettings(actorSystem, new StringDeserializer, new ByteArrayDeserializer)
      .withProperties(kafkaProperties)
val source =     Consumer.committableSource(consumerSettings, Subscriptions.topics(topics.toSet)).map { msg =>
      receivedCounter.foreach(_.inc())
      msg
    }

   source
      .mapAsync(maxParallel)(process)
      .mapAsync(maxParallel)(commit).runWith(Sink.ignore)

 def commit(committableMessage: CommittableMessage[String, Array[Byte]]): Future[Done] = {
      errorReporterWaitStrategy.reset()
      processorWaitStrategy.reset()
      committableMessage.committableOffset.commitScaladsl()
    }

def process(msg: MSG): Future[MSG] = {
      receivedCounter.foreach(_.inc())
      innerInvoke(msg)
    }

这是从我的代码的不同部分复制的片段。

scalaapache-kafkaakka-stream

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com