集合中 put/get 中的 Java 并发

Question

我通过在不同线程上运行的交互式会话 + 私有源 (InputStream) 连接到外部服务。在交互式会话中，我发送传出消息并接收包含不同字段的对象的同步响应，其中一个是 ID 和确认成功或失败的“状态”。同时，我在此 ID 的私人订阅源上收到消息，并带有进一步的“状态”更新。我目前将有关每个 ID 的状态信息存储在 ConcurrentHashMap 中。我必须在这些对象上保持正确的事件序列，但我目前遇到了竞争条件，有时我会在接收和处理交互式会话上的同步响应之前处理和更新私人提要上的对象，因此让我ID 的状态已过时且不正确。

理想情况下，我希望有某种类型的带有 PutIfKeyExistOrWait (w timeout) 方法的集合，它只会在密钥存在或等待时更新值，我可以在处理私有提要上的对象时使用它。

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

Answer 1

您可以尝试将处理这种情况的逻辑封装到地图的值中，如下所示：

• 如果 feed 线程是第一个为特定 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();
        }
    }
    ...
}

Answer 2

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

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

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

我不确定pendingFeedUpdates 使用什么同步数据结构：您需要按 ID 检索，但每个 ID 可能有很多消息，并且您希望保持消息的顺序。也许是一个同步的 HashMap，将每个 ID 与某种类型的同步有序队列相关联？

Answer 3

我建议您查看 Collections.getSynchronized 集合：http://docs.oracle.com/javase/1.4.2/docs/api/java/util/Collections.html#synchronizedList%28java.util.List %29

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

第三个选项是根据您要实现的目标，采用乐观或悲观的方法在应用程序中实施并发管理控制。这是 3 个中最复杂的一个，但如果与之前的选项结合使用，您将获得更大的控制权。

这真的取决于你的具体实现。