多线程等待批处理操作

Question

我有一个将多个任务提交到 ExecutorService 的进程，比如 MyTask。我的任务请求来自外部服务的值，例如 ExternalService。我想要做的是让 ExternalService 批处理请求，比如每 100 个传入或每 1 秒一次，直到那个时候让 MyTask 线程暂停，直到他们得到答案：

public class MyTask implements Runnable {

    @Override
    public void run() {
        try {
            // .... code ....


            ExternalData data = externalService.getData(id);

            // ..... code after batch ...

        }catch (Exception e){

        }
    }
}

我需要服务 externalService 暂停 MyTask 直到它批量操作（通过累积 100 个请求或等待 1 秒）并返回结果到任务继续。

解决这个问题的最佳方法是什么？

谢谢

Answer 1

这可以通过使用CyclicBarrier 来完成。

CyclicBarriers 在涉及固定大小的线程组的程序中很有用，这些线程组必须偶尔相互等待。

创建一个所有任务都可以访问的屏障。参数是调用await()时要等待的任务数量以及等待后要执行的操作：

CyclicBarrier barrier = new CyclicBarrier(100, action);

每个任务都应该调用await。这将等到总共 100 个任务达到这一点或最多 1 秒：

barrier.await(1, TimeUnit.SECONDS);

如果超时，无论如何都运行action。 action 是一个调用外部服务的可运行对象。由于action 并发运行，您可能还需要让您的任务等待它终止并获得结果。

---

要获得外部服务的实际结果，我可能会使用如下内容。基本上将 ID 保存在列表中并使其阻塞，直到结果准备好。

action.addId(id);
barrier.await(...); // run batch action using all IDs
result = action.getResult(id); // blocks until result is ready