3

我正在使用Monix Task进行异步控制。

设想

  1. 任务并行执行
  2. 如果故障发生 X 次以上
  3. 停止所有尚未完成的任务（越快越好）

我的解决方案

我提出了在 1. 结果和 2. 错误计数器之间竞争的想法，并取消失败者。
通过Task.race如果错误计数器首先达到阈值，那么任务将被取消Task.race。

实验

关于菊石REPL

{
  import $ivy.`io.monix::monix:3.1.0`
  import monix.eval.Task
  import monix.execution.atomic.Atomic
  import scala.concurrent.duration._
  import monix.execution.Scheduler
  //import monix.execution.Scheduler.Implicits.global
  implicit val s = Scheduler.fixedPool("race", 2) // pool size

  val taskSize = 100
  val errCounter = Atomic(0)
  val threshold = 3

  val tasks = (1 to taskSize).map(_ => Task.sleep(100.millis).map(_ => errCounter.increment()))
  val guard = Task(f"stop because too many error: ${errCounter.get()}")
    .restartUntil(_ => errCounter.get() >= threshold)

  val race = Task
    .race(guard, Task.gather(tasks))
    .runToFuture
    .onComplete { case x => println(x); println(f"completed task: ${errCounter.get()}") }
}

问题

结果取决于线程池大小！？

对于池大小 1
，结果几乎总是任务成功，即没有停止。

Success(Right(.........))
completed task: 100 // all task success !

对于池大小 2
，成功和失败之间的不确定性非常不确定，并且取消也不准确。例如：

Success(Left(stop because too many error: 1))
completed task: 98

取消最晚至 98 个任务已完成。
错误计数很奇怪，小到阈值。

默认全局调度程序获得相同的结果行为。

对于池大小 200
，它更具确定性，并且停止更早，因此在完成更少任务的意义上更准确。

Success(Left(stop because too many error: 2))
completed task: 8

池大小越大越好。

如果我Task.gather改为Task.sequence执行，所有问题都消失了！

这种依赖池大小的原因是什么？一旦发生太多错误，如何改进它或者是否有更好的选择来停止任务？

4

1 回答 1

1

您所看到的可能是 monix 调度程序的效果以及它旨在实现公平性的方式。这是一个相当复杂的话题，但文档和 scaladocs 非常好（参见：https ://monix.io/docs/3x/execution/scheduler.html#execution-model ）

当您只有一个线程（或几个线程）时，“守卫”任务需要一段时间才能进行另一轮检查。您一次启动 100 个任务，因此Task.gather调度程序非常忙，“守卫”在其他任务完成之前无法再次检查。如果每个任务有一个线程，则调度程序无法保证公平性，因此“守卫”会更频繁地进行不公平的检查，并且可以更快地完成。

如果您使用Task.sequence这 100 个任务按顺序执行，这就是“守卫”任务有更多机会在需要时立即完成的原因。如果您想保持代码原样，您可以使用Task.gatherN(parallelism = 4)which 将限制并行性，因此允许您的“守卫”更频繁地检查（和之间的中间地带Task.sequence）Task.gather。

对我来说，这似乎有点像 Go 代码（使用Task.race类似 Go's select），而且您还使用不受约束的副作用，这进一步使理解正在发生的事情变得复杂。我尝试以一种更惯用的方式重写您的程序，对于复杂的并发，我通常会使用以下流Observable：

import cats.effect.concurrent.Ref
import monix.eval.Task
import monix.execution.Scheduler
import monix.reactive.Observable

import scala.concurrent.duration._

object ErrorThresholdDemo extends App {

  //import monix.execution.Scheduler.Implicits.global
  implicit val s: Scheduler = Scheduler.fixedPool("race", 2) // pool size

  val taskSize  = 100
  val threshold = 30

  val program = for {
    errCounter <- Ref[Task].of(0)

    tasks = (1 to taskSize).map(n => Task.sleep(100.millis).flatMap(_ => errCounter.update(_ + (n % 2))))

    tasksFinishedCount <- Observable
      .fromIterable(tasks)
      .mapParallelUnordered(parallelism = 4) { task =>
        task
      }
      .takeUntilEval(errCounter.get.restartUntil(_ >= threshold))
      .map(_ => 1)
      .sumL

    errorCount <- errCounter.get
    _          <- Task(println(f"completed tasks: $tasksFinishedCount, errors: $errorCount"))
  } yield ()

  program.runSyncUnsafe()
}

正如你所看到的，我不再使用全局可变副作用，而是Ref内部也使用Atomic但提供了一个我们可以使用的功能性 api Task。出于演示目的，我还将阈值更改为 30，只有其他所有任务都会“出错”。completed tasks: 60, errors: 30因此，无论线程池大小如何，预期的输出总是在附近。

我仍在使用轮询，errCounter.get.restartUntil(_ >= threshold)这可能会根据我的口味消耗过多的 CPU，但它接近您的原始想法并且效果很好。

通常我不会预先创建任务列表，而是将输入放入 Observable 并在.mapParallelUnordered. 此代码保留您的列表，这就是为什么不涉及真正的映射（它已经包含任务）。

你可以选择你想要的并行度，就像Task.gatherNimo 非常棒一样。

如果还有什么不清楚的，请告诉我:)

于 2020-02-18T17:37:42.463 回答