如何在Akka调度程序中使用自定义执行程序 [英] How to use custom executor in Akka dispatcher

问题描述

我正在尝试在Akka中为调度程序使用自定义执行程序。

I am trying to use a custom executor for a dispatcher in Akka.

特别是，我想用我自己的包装现有执行程序（记录，调试等）。 。

Specifically, I want to wrap an existing executor with my own (logging, debugging, etc.).

我看过文档：

可以配置默认调度程序，默认情况下是具有 fork-join-executor的分派器，在大多数情况下，该分派器均具有出色的性能。

The default dispatcher can be configured, and is by default a Dispatcher with a "fork-join-executor", which gives excellent performance in most cases.

和

使用执行器指定，使用 fork-join-executor， thread-pool-executor或akka.dispatcher.ExecutorServiceConfigurator的FQCN

specify using "executor" using "fork-join-executor", "thread-pool-executor" or the FQCN of an akka.dispatcher.ExecutorServiceConfigurator

（我不知道什么是FQCN。）

(I don't understand what FQCN is.)

我该怎么办

推荐答案

FQCN ==完全合格的类名，即包名+类名，例如：

FQCN == Fully Qualified Class Name i.e. package name(s) + class name, ex:

java.lang.String< -字符串的FQCN

java.lang.String <- FQCN for String

以下是内置ThreadPoolExecutor的ExecutorServiceConfigurator的示例：

Here's an example of an ExecutorServiceConfigurator for the built in ThreadPoolExecutor:

class ThreadPoolExecutorConfigurator(config: Config, prerequisites: DispatcherPrerequisites) extends ExecutorServiceConfigurator(config, prerequisites) {

  val threadPoolConfig: ThreadPoolConfig = createThreadPoolConfigBuilder(config, prerequisites).config

  protected def createThreadPoolConfigBuilder(config: Config, prerequisites: DispatcherPrerequisites): ThreadPoolConfigBuilder = {
    import akka.util.Helpers.ConfigOps
    ThreadPoolConfigBuilder(ThreadPoolConfig())
      .setKeepAliveTime(config.getMillisDuration("keep-alive-time"))
      .setAllowCoreThreadTimeout(config getBoolean "allow-core-timeout")
      .setCorePoolSizeFromFactor(config getInt "core-pool-size-min", config getDouble "core-pool-size-factor", config getInt "core-pool-size-max")
      .setMaxPoolSizeFromFactor(config getInt "max-pool-size-min", config getDouble "max-pool-size-factor", config getInt "max-pool-size-max")
      .configure(
        Some(config getInt "task-queue-size") flatMap {
          case size if size > 0 ⇒
            Some(config getString "task-queue-type") map {
              case "array"       ⇒ ThreadPoolConfig.arrayBlockingQueue(size, false) //TODO config fairness?
              case "" | "linked" ⇒ ThreadPoolConfig.linkedBlockingQueue(size)
              case x             ⇒ throw new IllegalArgumentException("[%s] is not a valid task-queue-type [array|linked]!" format x)
            } map { qf ⇒ (q: ThreadPoolConfigBuilder) ⇒ q.setQueueFactory(qf) }
          case _ ⇒ None
        })
  }

  def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory =
    threadPoolConfig.createExecutorServiceFactory(id, threadFactory)
}

来源： https://github.com/akka/akka/blob/v2.3.4/akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala#L344

这篇关于如何在Akka调度程序中使用自定义执行程序的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！