Akka 流：Alpakka 不使用多个 CPU 内核

问题描述

我们创建了一个 Alpakka 流，它使用来自主题的 Kafka 消息，然后处理这些消息。这些消息是并行处理的，使用 mapAsyncUnordered 和配置的并行性。消费者的 Kafka 延迟增加，但应用程序仅使用 1 个 cpu 核心。我已将默认调度程序更改为 akka.actor.default-dispatchers，它使用 fork-join 执行程序，期望它使用的不仅仅是 cpu 内核。我的应用程序在 32 个内核中运行。 请在下面找到配置的设置：

akka.kafka.consumer.use-dispatcher = "akka.actor.default-dispatcher"

消费流代码：

Consumer.DrainingControl<Done> control = Consumer.committableSource(consumerSettings,Subscriptions.topics(topic))

                .buffer( 500,OverflowStrategy.backpressure() )

                //De-serialize the response from json to java object
                .mapAsyncUnordered( 5,//deserialize the output )

                .mapAsyncUnordered(5,//Process it and perform some calculations )

                .mapAsyncUnordered( 5,//Do something and return the consumer offset )

                //Commit the offset
                .toMat( Committer.sink(committerSettings.withMaxBatch(100)),Consumer::createDrainingControl)
                .run( materializer );

流在 akka-cluster 中运行，该集群由相同的消费者组 ID 进行负载平衡。我们在应用程序中也有一个类型化的actor系统，用于触发请求，以及一个有助于跨集群共享负载的组路由器。触发的请求作为 Kafka 消息发送到微服务，我们得到响应作为由流处理的 Kafka 消息。而且这些消息不一定要按顺序处理，因此使用mapAsyncUnordered...

尝试将并行度增加到 100，但没有看到任何变化。

提前致谢

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）