scala - Monix Task.sleep 和单线程执行

Question

我正在尝试理解 Monix 中的任务调度原则。 以下代码(来源:https://slides.com/avasil/fp-concurrency-scalamatsuri2019#/4/3)按预期仅生成“1”。

  val s1: Scheduler = Scheduler(
    ExecutionContext.fromExecutor(Executors.newSingleThreadExecutor()),
    ExecutionModel.SynchronousExecution)

  def repeat(id: Int): Task[Unit] =
    Task(println(s"$id ${Thread.currentThread().getName}")) >> repeat(id)

  val prog: Task[(Unit, Unit)] = (repeat(1), repeat(2)).parTupled

  prog.runToFuture(s1)

  // Output:
  // 1 pool-1-thread-1
  // 1 pool-1-thread-1
  // 1 pool-1-thread-1
  // ...

当我们在repeat方法中添加Task.sleep

  def repeat(id: Int): Task[Unit] =
    Task(println(s"$id ${Thread.currentThread().getName}")) >>
      Task.sleep(1.millis) >> repeat(id)

输出变为

// Output
// 1 pool-1-thread-1
// 2 pool-1-thread-1
// 1 pool-1-thread-1
// 2 pool-1-thread-1
// ...

这两个任务现在在一个线程上同时执行!好的 :) 一些合作屈服已经开始。这里究竟发生了什么？谢谢:)

编辑:Task.shift 而不是 Task.sleep。

Answer 1

我不确定这是否是您正在寻找的答案，但它是这样的:

尽管命名方式不同，Task.sleep 无法与更传统的方法(如 Thread.sleep)进行比较。

Task.sleep 实际上并不在线程上运行，而是简单地指示调度程序在耗时后运行回调。

这是来自 monix/TaskSleep.scala 的一小段代码比较:

[...]

implicit val s = ctx.scheduler
val c = TaskConnectionRef()
ctx.connection.push(c.cancel)

c := ctx.scheduler.scheduleOnce(
  timespan.length,
  timespan.unit,
  new SleepRunnable(ctx, cb)
)

[...]

private final class SleepRunnable(ctx: Context, cb: Callback[Throwable, Unit]) extends Runnable {

  def run(): Unit = {
    ctx.connection.pop()
    // We had an async boundary, as we must reset the frame
    ctx.frameRef.reset()
    cb.onSuccess(())
  }
}

[...]

在执行回调(此处:cb)之前，您的单线程调度程序(此处:ctx.scheduler)可以简单地使用他的线程进行任何操作计算排在下一个队列中。

这也解释了为什么这种方法更可取，因为我们不会在 sleep 间隔期间阻塞线程 - 浪费更少的计算周期。

希望这对您有所帮助。