我正在学习Reactor,我想知道如何实现特定的行为。 假设我有一系列传入的消息。 每个消息都与某个实体相关联并包含一些数据。
interface Message {
String getEntityId();
Data getData();
}
不同实体相关的消息可以并行处理。然而,任何单个实体的消息必须一次处理一个,即在实体 "abc"
的消息 1 处理完成之前不能开始处理实体 "abc"
的消息 2。
在处理消息时,应缓冲该实体的其他消息。
其他实体的消息可以无阻地进行。
可以将其视为每个实体有一个线程运行以下代码:
public void run() {
for (;;) {
// Blocks until there's a message available
Message msg = messageQueue.nextMessageFor(this.entityId);
// Blocks until processing is finished
processMessage(msg);
}
}
如何在React中实现非阻塞操作?总消息速率可能很高,但每个实体的消息速率将非常低。实体集可能非常大,并且不一定事先知道。
我猜它可能看起来像这样,但我不确定。
{
incomingMessages()
.groupBy(Message::getEntityId)
.flatMap(entityStream -> entityStream
/* ... */
.map(msg -> /* process the message */)))
/* ... */
}
public static Stream<Message> incomingMessages() { /* ... */ }