java - 并行和顺序的事件处理

Question

我正在设计一个java模块来处理其他外部系统发送的事件。系统接收到的所有事件都会经过映射，然后排队到 ThreadPoolExecutor 中进行进一步处理。

相当简单。

现在，这里的问题是，如果我收到具有相同键(例如“SystemID + EventEntityID”)的事件，我必须按顺序处理它们。由于事件频率为中等到高，我无法按顺序处理所有事件。这是毫无疑问的。

因此，当我收到一个事件时，我需要确保任何具有相同键的事件不在处理中或在队列中，然后再将其插入队列。另外，如果我发现它正在处理或在队列中，我必须将其保存在某个地方，然后在现有处理完成后将其推送到队列。

我不确定这个解决方案的良好设计方法是什么。任何帮助都会非常有帮助。

Answer 1

据我了解，事件的流程是:接收 -> 映射到队列 -> 执行。现在可以有很多解决方案，具有不同的效果。因为我实际上并不了解你们的系统，所以我提供的可能不是最合适的，只是一个基本的实现方向。

为了实现您所需要的，我们实际上需要修改的是队列。因此，我们将创建一个支持此功能的队列实现。

通常，当我遇到可以匹配多个值的键的问题时，我会使用 Map<K, Collection<V>>哪里K是 key 类型且 V是值类型。当然这不是一个队列，但它将用作我们队列实现的一部分。

当添加新事件时，我们可以将其放入与键匹配的映射中的集合中。从那里可以对其进行排队。

需要优先级排队，因此我们将使用一种PriorityQueue (实现可能会改变)。

我们需要注意几件事:

• 将新事件接收到 map 中。这个比较简单，放入 map 的集合值即可。
• 从 map 转移到队列。我们需要将每个键的集合中的第一个值放入优先级队列中。然而，这只能在处理完最后一个值后才能完成。
• 从队列轮询到用户。只需从优先级队列中检索第一个值即可。

将新事件接收到 map

这是在我们的推送方法中完成的，该方法将接收一个事件，并且非常简单:

Collection<V> collection = map.get(key);
if (collection == null) {
    collection = (create collection)
    map.put(key, collection);
}
collection.add(value);

从 map 转移到队列

这是最有问题的，因为我们需要找到一种方法来知道每种类型的事件是否在队列中以及最后一个事件何时被处理。我的第一个想法是使用代理事件值。该值将包含该数据，并且看起来像这样:

interface StoredEventType {
    // when a new event of this type was received
    void registerNewEvent(Event e);
    // move from map to queue
    void pollNextEventInto(Queue<StoredEvent> q);
}
interface StoredEvent {
    // when the execution was done for this event, this will be called.
    boolean isDone();
    // the execution should call this to update that it is done.
    void onDone();
}

只有当 isDone 时才会发生这种情况被调用时，会在 pollNextEventInto 时将一个新事件放入队列中叫做。为此，执行任务必须调用 onDone在最后。

什么时候 pollNextEventInto叫？当我们的队列实现被轮询时，可能应该调用它。这是执行器运行时唯一保证持续发生的事情。

从队列轮询到用户

这也很简单。首先执行pollNextEventInto在所有存储的键上，以确保更新优先级队列。然后简单地从优先级队列中轮询。

实现摘要

这不是一个完美的想法。必须保证它的并发性，抽象并不是最好的。但它给出了如何做的方向。另请注意 poll 中的大部分工作是如何完成的。 。这是因为poll在内部用于执行，而 push更加外部化，以某种方式与其他系统进行交互。

如果您想对此进行更多的集思广益，我很乐意。

编辑

我有机会坐下来检查这个实现。效果很好。不是最干净的代码...尽管经过一些努力它可以是。

事件队列实现可能应该支持阻塞 poll对于从中轮询并执行的线程。

运行的时候，我一直依赖Future检查事件处理是否完成。它对于异步工作非常有用。