multithreading - ExecutionContext.global和主线程之间的区别

Question

我有以下代码:

object KafkaApi {

  private implicit val main: ExecutionContextExecutor = ExecutionContext.global
  private val workers = ExecutionContext.fromExecutor(Executors.newCachedThreadPool())

  def main(args: Array[String]) {

    foo.unsafeRunAsync(_ => ())
    //foo.unsafeRunSync()
    println("Hello")


  }

  def foo: IO[Unit] =
    for {
      _ <- IO {
        println(Thread.currentThread().getName)
      }
      _ <- IO.shift(workers)
      _ <- IO {
        println(Thread.currentThread().getName)
      }
      _ <- IO {
        println(Thread.currentThread().getName)
      }
      _ <- IO {
        println(Thread.currentThread().getName)
      }
      _ <- IO.shift(main)
      _ <- IO {
        println(Thread.currentThread().getName)
      }
      _ <- IO {
        println(Thread.currentThread().getName)
      }
      _ <- IO {
        println(Thread.currentThread().getName)
      }
    } yield ()
}

输出为:

main
Hello
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
scala-execution-context-global-14
scala-execution-context-global-14
scala-execution-context-global-14

main 和 scala-execution-context-global-14 有什么区别？

如果这两个不同，如何恢复主线程？

Answer 1

Running the code above, why the application never get terminated?

这个额外的问题太大了，无法发表评论，因此我在这里回答。

事实是，在JVM中，所有 Thread都分为“普通”和 "daemon"线程。这里重要的是

The Java Virtual Machine exits when the only threads running are all daemon threads.

因此，如果您有任何正在运行的非守护进程 Thread，则JVM会认为您的应用程序仍在工作，即使它实际上什么也不做(也许只是在等待一些输入)。 “主”线程显然是“普通”线程。由标准 ExecutionContext.global创建的线程是守护程序，因此不会在主线程完成后阻止您的应用程序退出。由Java的 Executors.newCachedThreadPool创建的线程是非守护程序，因此可以使应用程序保持 Activity 状态。有几种可能的解决方案:

除了ExecutionContext外，请勿使用其他global，即完全不使用Executors.newCachedThreadPool。根据您的情况，这可能是您想要的，也可能不是。

完成所有工作后，显式 shutdown 您的自定义ExecutorService。这里要小心，因为shutdown不会等待所有 Activity 任务完成。所以代码应该变成像

private val pool = Executors.newCachedThreadPool
implicit private val workers = ExecutionContext.fromExecutor(pool)

// do whatever you want with workers 


// optionally wait for all the work to be done

pool.shutdown()

使用创建守护程序线程的自定义池。例如，您可以执行以下操作:

val workers = ExecutionContext.fromExecutor(Executors.newCachedThreadPool(new ThreadFactory {
  private val defaultDelegate = Executors.defaultThreadFactory()

  override def newThread(r: Runnable): Thread = {
    val t = defaultDelegate.newThread(r)
    //let the default factory do all the job and just override daemon-flag 
    t.setDaemon(true)
    t
  }
}))

恕我直言，＃2和＃3之间的主要 权衡是便利性与正确性。在＃3中，您不必考虑所有任务的完成位置，因此可以安全地调用 shutdown，这很方便。代价是，如果由于某种原因您在执行其他所有任务之前判断错误并退出了“主”线程，您将不会知道任何错误，因为守护程序线程将被静默杀死。如果您使用＃2并发生相同的错误，或者您未在该代码路径中调用 shutdown，则您的应用将继续运行，或者您会在日志中看到一些警告，指出在有一些任务的情况下池已关闭进行中。因此，如果这只是一长串处理的​​中间步骤，出于某种原因需要自定义线程池，那么我可能会选择＃3；但是如果这种并行执行是主要行为，那么我将采用更明确的＃2方法。