处理大量有状态事件的 Java 最佳实践

Question

我正在寻找实用程序类或最佳实践模式来处理我的应用程序中的大量传入有状态事件。

想象一个生产者产生许多事件，然后由作用于这些事件的应用程序使用。现在在某些情况下，生产者产生的事件比消费者实际处理的要多，但是由于所有事件都是有状态的，因此是否会错过某些事件并不重要，因为最新事件包含先前事件传达的所有信息。

我现在编写了以下 java 代码来处理这些情况，但我不确定这是否是正确的方法，以及是否没有更简单、更好、更安全的方法。

private static ScheduledThreadPoolExecutor executorService = new ScheduledThreadPoolExecutor(1);
private final static Object lock = new Object();
private static List<EventData> lastEventData = null;

static {
    executorService.scheduleWithFixedDelay(new Runnable() {
        @Override
        public void run() {
            synchronized(lock) {
                while(lastEventData == null && !executorService.isShutdown()) {
                    try {
                        lock.wait();
                    } catch (InterruptedException ex) { ... }
                }
                try {
                    actUponEvent(lastEventData);
                } catch (Throwable ex) { ... }
                lastEventData = null;
            }
        }
    }, 250, 250, TimeUnit.MILLISECONDS);
}


public synchronized update(final List<EventData> data) {
    synchronized(lock) {
        lastEventData = data;
        lock.notifyAll();
    }
}

public void dispose() {
    executorService.shutdown();
}

换句话说，我想在事件通知到达后立即收到通知，但将它们限制为每 250 毫秒一个事件，我只对最后一个传入事件感兴趣。

我通过 java.util.concurrent 查看了一些提示/现有解决方案，但找不到任何适合我的问题的东西。 BlockingQueue 起初似乎非常好，因为如果为空，它会阻塞，但另一方面，队列本身对我来说并不重要，因为无论如何我只对最新事件感兴趣，如果满了则插入阻塞不是我也在寻找什么。

Answer 1

以下模型可以支持非常高的更新率，（达到每秒数千万）但您只需要在内存中保留最新的。

---

如果您每 N 毫秒拍摄一次快照，您可以使用这种方法。

final AtomicReference<ConcurrentHashMap<Key, Event>> mapRef =

当您有更新时，将其添加到 ConcurrentMap。选择键以便应该替换前一个事件的事件具有相同的键。

Key key = keyFor(event);
mapRef.get().put(key, event);

这种映射方式随时都有任何键的最新更新。

有一个每 N 毫秒运行一次的任务。此任务在运行时可以将地图换成另一个（或以前的空地图以避免创建新地图）

ConcurrentMap<Key, Event> prev = mapRef.set(prevEmptyMap);

for(Event e: prev.values())
    process(e);
prev.clear();
this.prevEmptymap = prev;