集合中 put/get 中的 Java 并发

Question

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

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

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

Answer 1

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

• 如果 feed 线程是第一个为特定 id 添加值的线程，则该值被认为是不完整的，线程将等待直到它完成
• 如果交互式 session 线程不是第一个添加值的线程，它会将不完整的值标记为已完成
• 不完整的值在从 map 中获取时被视为不存在

此解决方案基于 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();
        }
    }
    ...
}