events - 如何使用 Reactor 向多个消费者派发事件，并根据事件数据过滤事件？

Question

我正在评估 Reactor ( https://github.com/reactor/reactor) 是否适合在我的 Spring/企业应用程序中创建事件调度框架。

首先，考虑一个场景，您有一个接口(interface) A和具体事件类 B , C ， 等等。我想将具体事件分派(dispatch)给多个消费者，即观察者。这些已注册到全局 Reactor bean 后期处理期间的实例。但是，您可以动态注册它们。在大多数情况下，一个生产者以高速率向多个消费者发送事件。

我用过Selector s，即 ClassSelector将正确的事件类型分派(dispatch)给正确的消费者。这似乎工作得很好。

Reactor reactor = ...
B event = ...
Consumer<Event<B>> consumer = ...

// Registration is used to cancel the subscription later
Registration<?> registration = reactor.on(T(event.getClass()), consumer);

要通知，使用事件的类型作为键

B event = ...
reactor.notify(event.getClass(), Event.wrap(event));

但是，我想知道这是否是建议的有效分派(dispatch)事件的方法？

其次，我想知道是否可以根据事件数据过滤事件？如果我理解正确，Selector s 仅用于检查 key 。我在这里不是指事件 header ，而是指域特定的对象属性。我想知道使用 Streams和 Stream.filter(Predicate<T> p)为此，但也可以使用 Reactor 进行过滤吗？和 Selector秒？当然，我可以编写一个委托(delegate)消费者来检查数据并在需要时将其委托(delegate)给注册消费者。

Answer 1

有一个名为 Selectors 的辅助对象这有助于创建各种内置 Selector实现。在那里您可以看到对 PredicateSelector 的引用. PredicateSelector非常有用，因为它允许您完全控制通知键的匹配。它可以是一个 Spring @Bean 、匿名内部类、lambda 或任何其他符合简单 Predicate 的东西界面。

可选，如果您有 JsonPath library在你的类路径中，然后你可以使用 JsonPathSelector 根据 JsonPath 查询进行匹配。

在这两种情况中的任何一种情况下，如果重要数据实际上是域对象本身，则您不需要为键设置单独的对象。只需通知对象并传递 Event<Object>作为第二个参数。

MyPojo p = service.next();
reactor.notify(p, Event.wrap(p));