java - 算法 - 仅针对队列中的唯一条目执行任务，常见条目应该等待

Question

我们正在创建一个休息应用程序。我们有一个边缘条件，即同一对象上不支持并行操作。例如:

Not supported in parallel
    Request 1 for action XYZ for object  A
    Request 2 for action XYZ for object  A
    Request 3 for action ABC for object  A

Supported in parallel
    Request 1 for action XYZ for object  A
    Request 2 for action XYZ for object  B
    Request 3 for action ABC for object  C

现在，对象计数不固定。我们可以有 n 个这样的对象。

我希望如果对象 A 的请求正在进行中，则对象 A 的其他请求应该等待对象 A 上的现有任务完成。

但我无法找出用于此目的的算法。

我可以规划以下设计，但无法弄清楚如何使用锁定，因为所有对象都可能不同。

• 当请求到来时存储对象 A 条目的队列。
• 如果发送回复，条目将被删除
• 如果条目已存在，则等待现有请求结束。
• 如果条目不存在，则立即执行。

现在对象 A 上的任务不应影响对象 B 上的任务。因此它们必须接受唯一锁。

而且，请求不能独立运行并排队。不知何故，我必须让当前线程 hibernate ，以便我可以向用户发送响应。

谁能指导一下吗？

Answer 1

根据我最初回复的评论进行更新

类似情况的理想模型是使用 Akka 等 Actor 系统。

但是您的评论指出，这将发生在 REST 应用程序的上下文中，其中线程将被请求处理阻塞。

在这种情况下，我们的想法是使用每个对象的防护，例如:

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;

public class ObjectGuard<K> {
    private final ConcurrentMap<K, CountDownLatch> activeTasks = new ConcurrentHashMap<>();

    public Guard guardFor(final K key) throws InterruptedException {
        if (key == null) {
            throw new NullPointerException("key cannot be null");
        }

        final CountDownLatch latch = new CountDownLatch(1);

        while (true) {
            final CountDownLatch currentOwner = activeTasks.putIfAbsent(key, latch);
            if (currentOwner == null) {
                break;
            } else {
                currentOwner.await();
            }
        }

        return () -> {
            activeTasks.remove(key);
            latch.countDown();
        };
    }

    public interface Guard extends AutoCloseable {
        @Override
        void close();
    }
}

您可以按如下方式使用它:

class RequestProcessor {
    private final ObjectGuard<String> perObjectGuard = new ObjectGuard<>();

    public String process(String objectId, String op) throws InterruptedException {
        // Only one thread per object id can be present at any given time
        try (ObjectGuard.Guard ignore = perObjectGuard.guardFor(objectId)) {
            String result = ... // compute response
        }
    }
}

如果针对同一对象 ID 接收到对 process 的两个并发调用，则只会处理一个，其他的则等待轮到处理该对象上的请求。