如何处理需要组合以创建响应的多个客户端请求？

Question

我正在使用 JAX-RS 来处理客户端请求。场景是这样的：客户端被分成 5 个组。每个客户端需要向服务器提交一个数字，当属于同一组的所有 5 个客户端完成提交他们的数字时，服务器计算这些数字的总和并返回结果给他们所有人。

因为当客户端提交一个数字时，他们必须等待其他人也这样做（如果他们不是最后一个）我正在使用异步响应 API，它允许等待其他客户端而不会超载服务器.假设没有超时限制。

我对 JAX-RS 的内部工作原理、会话和上下文缺乏了解，这让我很难设计出合适的解决方案。这是我的想法（混合代码-伪代码）：

public class SummationResource {

    @Inject
    SummationService service; // talks with the entity manager

             //group    client  number
    static Map<Long, Map<Long, Integer>> groupMap;

    @POST
    @Path("{clientId}")
    public void submit(int num,
                        @PathParam("clientId") long clientId,
                        @Suspended AsyncResponse response) {
        new Thread() {
            @Override
            public void run() {
                long groupId = service.getGroupOfClient(clientId);
                Map<Long, Integer> clients = groupMap.get(groupId);
                clients.put(clientId, num);
                if (clients.size() != 5)
                    // wait (see points below)
                else
                    // release all threads of this group
                int sum = HelperClass.sum(clients.values()); // should be executed only once
                response.resume(sum);
            }
        }.start();
    }
}

我想触摸的点：

• 我知道我可以将它存储在数据库中而不是内存中，但是假设客户端之间的提交时间很短，并且从数据库中写入和读取的成本更高。 groupMap 对于提交的每次调用是否不同（由不同的客户端）？使其静态足以让它在所有实例之间共享？
• 为了同步线程，我可以创建一个Map<Long, Object> locks，它将一个组ID 映射到一个同步的锁对象。所以每个线程通过调用locks.get(groupId).wait()来等待，释放行是locks.get(groupId).notifyAll()。锁对象由属于组的第一个线程创建。我不知道这种方法对 JAX-RS 和 AsyncResponse 是否友好。等待通知今天也没有太多使用，因此在这种情况下为此目的工作的任何新 API 都是好的。也许是ExecutorService 和Future？
• 我不确定如何将这些数字组合一次并将结果返回给所有线程。通常我会join他们，但我不能在线程之外这样做，因为这会阻止同步响应，从而破坏整个想法。

我正在寻找一种能够解决上述请求和要点的设计，类似于我在代码中所做的。

Answer 1

1. groupMap 是静态的，因此只要应用程序存在，它对于所有调用都是同一个实例。但是还没有创建实例。 Java 提供了同步集合，因此您无需费心同步对它的访问：

   static Map<Long, Map<Long, Integer>> groupMap = Collections.synchronizedMap( new HashMap<>());
   

并且您应该至少用您的客户填充一次（当前客户将为空）。

1. 对于等待，我建议使用 CompletableFuture，它可以返回自定义类型的结果，为您的每个 groupId 使用一个：

   Map<Long, CompletableFuture> futures = Collections.synchronizedMap( new HashMap<>());
   

每个必须等待的线程 (clients.size()

    CompletableFuture<Integer> future = new CompletableFuture<>();
    futures.put(clientId, future);
    int sum = future.get();

计算总和的第 5 个线程将完成所有这些期货，传递计算的总和。

    futures.values().forEach( i -> i.complete( sum));