项目反应堆处理器 v3.X

Question

我们正在尝试从 2.X 迁移到 3.X。 https://github.com/reactor/reactor-core/issues/375 我们在我们的应用程序（低延迟 FX 系统）中使用了 EventBus 作为事件管理器，它非常适合我们。

更改后，我们决定采用每个模块并创建自己的处理器来处理事件。 1. 从您的角度来看，这种用法是否正确？因为在当前阶段缺乏文件，并且在审查了所有我们可以做的事情之后，我们真的不知道该怎么做 2. 我们尝试使用 Flux 来执行每 X 间隔的动作 例如：市场在 1 秒内到达 1000，但我们希望在一秒钟内只处理 4 次更新。升级后我们正在使用：

带有缓冲区的处理器并发送到另一个方法。 在这种方法中，我们有 Flux 获取列表并尝试并行工作以完成他的任务。 我们有两个主要问题： 1. 有时我们收到 Null 事件，但我们无法发现系统正在发送给我，我想也许我们错过了使用处理器

    //Definition of processor
    ReplayProcessor<Event> classAEventProcessor = ReplayProcessor.create();

    //Event handler subscribing
    public void onMyEventX(Consumer<Event> consumer) {
       Flux<Event> handler = classAEventProcessor .filter(event -> event.getType().equals(EVENT_X));
       handler.subscribe(consumer);
   }

在上面的例子中，处理程序中的事件有时会为空。一旦他做了流停止工作，直到我们重新启动服务器（因为只有在重新启动时我们才会创建处理器）

2.我们尝试过并行，但有时一些消息消失了，所以也许我们滥用了框架

    //On constructor
   tickProcessor.buffer(1024, Duration.of(250, ChronoUnit.MILLIS)).subscribe(markets -> 
   handleMarkets(markets));

   //Handler
  Flux.fromIterable(getListToProcess())
        .parallel()
    .runOn(Schedulers.parallel())
    .doOnNext(entryMap -> {
        DoBlockingWork(entryMap);
    })
    .sequential()
    .subscribe();

这样做的目的是处理器将每 250 毫秒唤醒一次并调用处理程序。处理程序将与 Flux 并行工作，以进行更好和更快的处理。 *如果 DoBlockingWork 花费超过 250 毫秒，我无法理解会是什么行为

更新： EventBus 由我们包装，每个订阅的事件都会抛出包装的事件管理器。 现在我们尝试为每个模块创建事件处理器，但它的工作速度非常慢。我们已经将 TopicProcessor 与 ThreadExecutor 一起使用，但仍然很慢。EventBus 高速完成了同样的工作 有人有什么想法吗？顺便说一句，当我尝试使用 DirectProcessor 时，它似乎比 TopicProcessor 工作得更好

Answer 1

Reactor 3 是围绕您应尽可能避免阻塞的概念构建的，因此在您的第二个 sn-p DoBlockingWork 看起来不太好。

事件是如何产生的？您是否有基于侦听器的异步 API 来获取它们？如果是这样，您可以尝试使用Flux.create。

对于“我们在 1 秒内有 1000 个事件，但只想处理 4 个”的用例，我将链接一个 sample 运算符。例如，sample(Duration.ofMillis(250)) 将每秒分为 4 个窗口，它只会从中发出最后一个元素。

正在编写参考指南，以及一个可以找到外部文章和学习材料链接的页面。WIP 参考指南here 和学习资源页面here 的预览。