Java在并发中集合/获取集合

Question

我通过交互式会话+在独立线程上运行的私有源（InputStream）连接到外部服务。在交互式会话中，我发送传出消息并接收包含不同字段的对象的同步响应，其中一个是ID和“状态”，以确认成功或失败。同时，我会收到有关此ID的私人Feed的消息，并进一步提供“状态”更新。我目前在一个ConcurrentHashMap中存储关于每个ID的状态的信息。我必须在这些对象上保留正确的事件序列，但目前我正在获得竞争条件，在我接收和处理交互式会话上的同步响应之前，有时我会处理和更新专用Feed上的对象，因此让我与ID的过时和不正确的状态。

理想情况下，我希望拥有某种类型的带PutIfKeyExistOrWait（w超时）方法的集合，只有在密钥存在或等待时才会更新该值，以便在处理私有订阅源上的对象时使用。

有谁知道是否有合适的集合可用或可以建议我的问题的替代解决方案？谢谢。

Answer 1

您已经有一些ConcurrentHashMap iDAndStatus存储ID和最新状态。但是，我只会让处理该服务的线程在该地图中创建一个新条目。

当消息从Feed中到达时，如果ID已存在于iDAndStatus中，它只是修改状态。如果密钥不存在，只需将ID /状态更新临时存储在其他数据结构pendingFeedUpdates中。

每次在iDAndStatus中创建新条目时，请检查pendingFeedUpdates以查看是否存在新ID的某些更新。

我不确定用于pendingFeedUpdates的同步数据结构：您需要通过ID检索，但每个ID可能有许多消息，并且您希望保留消息的顺序。也许一个同步的HashMap将每个ID与某种类型的同步排序队列关联起来？

Answer 2

我建议你看一下Collections.getSynchronized集合：http：//docs.oracle.com/javase/1.4.2/docs/api/java/util/Collections.html#synchronizedList%28java.util。列表％29

这可能可能解决您的问题另一个选项取决于如何调用该方法是一种同步方法，它允许线程安全执行并确保事务的原子性。请参阅http://docs.oracle.com/javase/tutorial/essential/concurrency/syncmeth.html

第三种选择是根据您尝试实现的内容，在乐观或悲观方法之后，在应用程序内强制实施并发管理控制。这是3中最复杂的一个，但如果再加上以前的选项，会给你更大的控制权。

这实际上取决于您的具体实施。

Answer 3

你可以尝试封装逻辑处理这种情况到您的地图，像这样的价值观：

• 如果给的螺纹是第一个加入了特定的ID值，该值被认为是不完整的，线程等待，直到它完成
• 如果交互式会话线程不是第一次添加一个值，它标志着不完全值作为完全从地图

让它们时

不完整的值将被视为缺席

该解决方案基于putIfAbsent()的原子性。

public class StatusMap { 
    private Map<Long, StatusHolder> map = new ConcurrentHashMap<Long, StatusHolder>(); 

    public Status getStatus(long id) { 
     StatusHolder holder = map.get(id); 
     if (holder == null || holder.isIncomplete()) { 
      return null; 
     } else { 
      return holder.getStatus(); 
     } 
    } 

    public void newStatusFromInteractiveSession(long id, Status status) { 
     StatusHolder holder = StatusHolder.newComplete(status); 
     if ((holder = map.putIfAbsent(id, holder)) != null) { 
      holder.makeComplete(status); // Holder already exists, complete it 
     } 
    } 

    public void newStatusFromFeed(long id, Status status) { 
     StatusHolder incomplete = StatusHolder.newIncomplete(); 
     StatusHolder holder = null; 
     if ((holder = map.putIfAbsent(id, incomplete)) == null) { 
      holder = incomplete; // New holder added, wait for its completion 
      holder.waitForCompletion(); 
     } 
     holder.updateStatus(status); 
    } 
} 

public class StatusHolder { 
    private volatile Status status; 
    private volatile boolean incomplete; 
    private Object lock = new Object(); 

    private StatusHolder(Status status, boolean incomplete) { ... } 

    public static StatusHolder newComplete(Status status) { 
     return new StatusHolder(status, false); 
    } 

    public static StatusHolder newIncomplete() { 
     return new StatusHolder(null, true); 
    } 

    public boolean isIncomplete() { return incomplete; } 

    public void makeComplete(Status status) { 
     synchronized (lock) { 
      this.status = status; 
      incomplete = false; 
      lock.notifyAll(); 
     } 
    } 

    public void waitForCompletion() { 
     synchronized (lock) { 
      while (incomplete) lock.wait(); 
     } 
    } 
    ... 
}