如何在Reactor 3中跨多个发布服务器对工作进行排队？

Question

我正在创建一个库，用于使用Reactor3创建数据处理工作流。每个任务都有一个输入流量和一个输出流量。输入流量由用户提供。输出流量由库创建。任务可以链接以形成DAG。类似于这样：（它是静态编程语言）

val base64 = task<String, String>("base64") {
    input { Flux.just("a", "b", "c", "d", "e") }
    outputFn { ... get the output values ... }
    scriptFn { ... do some stuff ... }
}

val step2 = task<List<String>, String>("step2") {
    input { base64.output.buffer(3) }
    outputFn { ... }
    scriptFn { ... }
}

我需要限制整个工作流的并发性。一次只能处理配置数量的输入。在上面的示例中，对于3的限制，这意味着task base64将首先使用输入“a”、“b”和“c”运行，然后等待每个输入完成，然后再处理“d”、“e”和“step2”任务。

在从输入通量创建输出通量时，如何应用这些限制？TopicProcessor能否以某种方式应用？也许是某种定制的调度器或处理器？背压是如何工作的？我需要担心创建缓冲区吗？

Answer 1

背压从最后一个Susbuiber向上传播，穿过整个链条。但是链中的操作员可以提前请求数据（预取），甚至“重写”请求。例如，在缓冲区（3）的情况下，如果该操作员收到请求（1），它将执行上游请求（3）（“1缓冲区==最多3个元素，以便我可以请求足够的源来填充请求的1缓冲区”）。

如果输入总是由用户提供，这将很难抽象出来。。。

没有简单的方法可以对跨多个管道的源进行评级，甚至对给定管道的多个订阅进行评级（流量）。

在多个发布中使用共享的调度程序将不起作用，因为发布会选择一个工作线程并将其粘住。

但是，如果您的问题更具体地说是关于受限制的base64任务，那么可能可以从flatMap的并发参数中获得效果？

input.flatMap(someString -> asyncProcess(someString), 3, 1);

这将允许最多3次出现的asyncProcess运行，并且每次终止时，它都会从输入的下一个值开始一个新值。