java - 单个数据存储的多个消息监听器。高效设计

Question

我有一个由多个消息监听器写入的数据存储。每个消息监听器也可以位于数百个单独的线程中。

数据存储是一个PriorityBlockingQueue，因为它需要按时间戳对插入的对象进行排序。为了高效地检查项目队列而不是循环队列，使用并发 HashMap 作为索引形式。

private Map<String, SLAData> SLADataIndex = new ConcurrentHashMap<String, SLAData>();;
private BlockingQueue<SLAData> SLADataQueue;

问题 1 这是一个可接受的设计，还是我应该只使用单个 PriorityBlockingQueue。

每个消息监听器都会执行一个操作，这些监听器会扩展到多个线程。

插入方法，以便它插入到两者中。

this.SLADataIndex.put(dataToWrite.getMessageId(), dataToWrite);
this.SLADataQueue.add(dataToWrite);

更新方法

this.SLADataIndex.get(messageId).setNodeId(
            updatedNodeId);

删除方法

SLATupleData data = this.SLADataIndex.get(messageId);
//remove is O(log n)
this.SLADataQueue.remove(data);
// remove from index
this.SLADataIndex.remove(messageId);

问题二使用这些方法是最有效的方法吗？它们通过另一个对象有包装器来进行错误处理。

问题三 使用并发的 HashMap 和 BlockingQueue 是否意味着这些操作是线程安全的？我不需要使用锁对象？

问题四 当这些方法被多个线程和监听器调用且没有任何同步块(synchronized block)时，它们是否可以被不同的线程或监听器同时调用？

Answer 1

Question 1 is this a acceptable design or should I just use the single PriorityBlockingQueue.

当然，您应该尝试使用单个队列。保持两个集契约(Contract)步将需要更多的同步复杂性和代码中的担忧。

为什么需要 map ？如果只是调用 setNodeId(...) 那么我会让处理线程在从 Queue 中提取数据时自行执行此操作。

// processing thread
while (!Thread.currentThread().isInterrupted()) {
   dataToWrite = queue.take();
   dataToWrite.setNodeId(myNodeId);
   // process data
   ...
}

Question Two Using these methods is this the most efficient way? They have wrappers around them via another object for error handling.

当然，这看起来不错，但是，您再次需要进行一些同步锁定，否则您将遭受 race conditions 的困扰。保持两个集契约(Contract)步。

Question Three Using a concurrent HashMap and BlockingQueue does this mean these operations are thread safe? I dont need to use a lock object?

这两个类(ConcurrentHashMap 和 BlockingQueue 实现)都是线程安全的，是的。 但是因为有两个集合，所以您可能会遇到竞争条件，其中一个集合已更新，但另一个集合尚未更新。最有可能的是，您必须使用锁对象来确保两个集合正确保持同步。

Question Four When these methods are called by multiple threads and listeners without any sort of synchronized block, can they be called at the same time by different threads or listeners?

在没有看到相关代码的情况下，这是一个很难回答的问题。例如。当另一个线程调用 Delete(...) 并且该项目将在 Map 中找到并删除，但 queue.remove() 不会在队列，因为 Insert(...) 尚未在另一个线程中完成。