如何让线程/作业等到前一个作业完成，然后在某些通知上恢复它答案

【问题标题】：How to make Thread/Job wait until the previous Job is finished and thereafter resume it on some Notification如何让线程/作业等到前一个作业完成，然后在某些通知上恢复它
【发布时间】：2019-11-07 09:38:31
【问题描述】：

我有一个骆驼路线，它将对给定的 ID 执行更新。

对于单个请求，一切正常，例如 如果我有多个具有相同 ID 的请求，例如 10，那么我希望一次执行单个请求。

或等待其他具有相同ID的请求，同时如果我有其他ID的请求无需等待即可处理，例如12。

为此，我在我的 Camel 上下文“addRequestToPool”中创建了一个附加路由，这会将所有来自骆驼“updateTicket”的请求路由到 1 个地方，并从那里执行 1 对 1 的处理。

我想实现类似下面的东西：

当请求到达“synchronizeRequests”方法时，它应该检查是否已经存在带有 docketId 的作业，如果有的话，让它等到旧进程完成（比如 10）。如果没有这样的 id（比如 12）继续处理。

一旦旧进程完成，它应该告诉其他具有相同 ID 等待的作业，其中 1 个可以从它们等待的位置恢复工作，以便发送响应。

请求收集是从路由“addRequestToPool”完成的，一旦请求到来，它将为每个请求创建一个线程并调用同步方法。在同步方法“RequestSynchronization”内部，我想对具有 ID 的现有 Job 执行检查操作，如果存在则等待来自先前的通知，如果没有返回响应并继续处理。

一旦使用路由“removeRequestFromPool”完成作业，作业的当前 ID 将被删除，并且应该通知组中的新作业（相同 ID）继续。

我想知道我应该如何实施上述方案或任何其他适当的方法（如果有）？

以下是示例代码供参考，

骆驼路线：

from("direct:updateTicket")
    .to("direct:addRequestToPool")
    .to("direct:getDetailForTicket")
    .to("direct:updateDetailForTroubleTicket")
    .to("direct:getDetailForTicket")
    .to("direct:removeRequestFromPool")
    .to("direct:endRoute");

API 调用：

public class TicketRequestManagement {

    public SimpleActionResponse addRequestToPool(int docketNo) throws InterruptedException {

        SimpleActionResponse response = new SimpleActionResponse();

        RequestSynchronization requestSynchronization = new RequestSynchronization(docketNo);
        Thread thread = new Thread(requestSynchronization);
        thread.start();
        thread.join();
        return response;
    }
}

同步方式：

public class RequestSynchronization implements Runnable {

    private static Logger logger = Logger.getLogger(RequestSynchronization.class);

    private int docketNo;

    public RequestSynchronization(int docketNo) {
        super();
        this.docketNo = docketNo;
    }

    @Override
    public void run() {
        synchronizeRequests(docketNo);
    }

    public synchronized static void synchronizeRequests(int docketNo) {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

我正在考虑将 docketId 添加到静态数据结构（Set）并检查是否存在 Id 等待（我应该如何执行此等待？）一旦进程完成从集合中删除 ID 并通知等待进程从那里恢复（我该如何向同一个 ID 执行此通知？）。

如果需要任何澄清，请告诉我。

PS：我不是在找睡眠，这只是为了进行同步测试。

提前谢谢你。

【问题讨论】：

我不熟悉 apache-camel，但还是想让您了解 JDK 的并发实用程序。请参阅包java.util.concurrent 以及tech guide，最后是tutorial。当然，这些不是唯一可用的参考资料。 Google for java 并发实用程序。

标签： java multithreading apache-camel

【解决方案1】：

我的建议：去异步。例如，使用 JMS（不是直接使用，而是使用 Camels excellent JMS support）来分离请求和处理它们。通过这种方式，您可以简单地通过消费者数量来控制整个线程。

在你的情况下，这将是

addRequestToPool 将请求放在 JMS 队列中
JMS 消费者使用队列中的请求并处理它们
使用message groups同步请求同ID
使用消费者计数来确保并行处理其他 ID 并扩大规模（您编写 0 行代码来处理线程）
即使处理组件因维护而停机，客户端仍然可以发送更新请求

消息组是 JMS 代理功能（至少是 ActiveMQ）。您将 ID 设置为 JMSXGroupID，并且代理确保具有相同 ID 的所有消息都由同一消费者处理。如果您有 5 个消费者，并且您收到 10 条具有相同 ID 的消息，那么 1 个消费者会完成所有工作。一旦具有另一个 ID 的消息到达，另一个消费者就会并行处理它。

您是否构建了一个大型 JMS 组件来执行所有处理，或者您构建多个组件来通过处理工作流中的多个队列传递消息，这取决于您。

【讨论】：

【解决方案2】：

首先，不要手动启动Thread，这就是ExecutorServices 的用途。

其次，我建议您在 Map<Integer, Future<>> 中维护已启动的同步，以便检查该 ID 是否已经在运行。

草稿是

class TicketRequestService {
    private Map<Integer, Future<?>> runningRequests = new ConcurrentHashMap<>();
    ExecutorService executor = Executors.newCachedThreadPool();

    public void synchronizeRequest(int docketId) {
        // get the running Future if present, otherwise...
        Future<?> future = runningRequests.computeIfAbsent(docketId,
            // ...create a new one
            id-> executor.submit(() -> {
                RequestSynchronization.synchronizeRequests(id));
                // remove future from map after sync is finished
                runningRequests.remove(docketId);
            });
        future.get(); // wait until finished; this throws Exceptions, you need to handle them
        return new SimpleActionResponse();
   }
}

【讨论】：