Java中同步多个异步请求

Question

我正在使用 Java 中的官方 Telegram Api (TDLib) 来请求有关群组所有成员的信息。使用它们的 ID，我向服务器发送异步请求，并User在 ResultHandler 内接收每个请求的对象，如下所示：

private static ArrayList<TdApi.User> chatUsers= new ArrayList<>();

private static void addUsers(){

    for (int i = 0; i < userIDs.length; i++){

        client.send(new TdApi.GetUser(userIDs[i]), new Client.ResultHandler() {

                        @Override
                        public void onResult(TdApi.Object object) {
                            TdApi.User user = (TdApi.User)object;
                            chatUsers.add(user);
                        }
        });
    }
}

由于我对 Java 中的异步请求还很陌生，所以我想知道以下内容：

1. 调用此方法并在继续之前等待收到的所有结果的适当方法是什么？

2. 一般来说，当连续调用多个请求并在继续下一个请求之前等待每个结果时，除了将请求嵌套在彼此内部以在 Java 中同步之外，通常的方法是什么？我想避免这样的事情：

   private static void getSupergroupId(int chatId){
   
   //first step
   client.send(new TdApi.GetChat(chatId), new Client.ResultHandler() {
               @Override
               public void onResult(TdApi.Object object) {
                   supergroupId = ((TdApi.ChatTypeSupergroup)((TdApi.Chat)object).type).supergroupId;
   
                   //second step when result received
                   client.send(new TdApi.GetSupergroupMembers(supergroupId, null, 0, 200), new Client.ResultHandler() {
                       @Override
                       public void onResult(TdApi.Object object) {
                           chatMembers = ((TdApi.ChatMembers)object).members;
   
                           //further steps which need to wait for the result of the step before
                       }
                   });
               }
   });
   }
   

谢谢你！

Answer 1

1 Java 同步器之一应该可以工作。CountDownLatch我将从最简单的一个开始。

  private static final ArrayList<TdApi.User> chatUsers = Collections.synchronizedList(new ArrayList<>());

  private static void addUsers() {
    final CountDownLatch latch = new CountDownLatch(userIDs.length);
    for (int i = 0; i < userIDs.length; i++) {
      client.send(new TdApi.GetUser(userIDs[i]), new Client.ResultHandler() {
        @Override
        public void onResult(TdApi.Object object) {
          TdApi.User user = (TdApi.User) object;
          chatUsers.add(user);
          latch.countDown();
        }
      });
    }
    // handle InterruptedException
    latch.await(10, TimeUnit.SECONDS);
  }

请注意，它chatUsers是从不同的线程访问的，因此对它的访问应该受到锁的保护。为了简单起见，我Collections.synchronizedList在示例中使用了。但是，您应该使用更细粒度的方法。

2 看看Completablefuture，好像就是你要找的。

  private static void getSupergroupId(int chatId) {
    CompletableFuture.supplyAsync(() -> {
      AtomicReference<TdApi.ChatTypeSupergroup> atomicReference = new AtomicReference<>();
      CountDownLatch latch = new CountDownLatch(1);
      client.send(new TdApi.GetChat(chatId), new Client.ResultHandler() {
        @Override
        public void onResult(TdApi.Object object) {
          atomicReference.set(((TdApi.ChatTypeSupergroup) ((TdApi.Chat) object).type).supergroupId);
          latch.countDown();
        }
      });
      // handle InterruptedException
      latch.await(10, TimeUnit.SECONDS);
      return atomicReference.get();
    }).thenApply(supergroupId -> {
      AtomicReference<TdApi.ChatMembers> atomicReference = new AtomicReference<>();
      CountDownLatch latch = new CountDownLatch(1);
      client.send(new TdApi.GetSupergroupMembers(supergroupId, null, 0, 200), new Client.ResultHandler() {
        @Override
        public void onResult(TdApi.Object object) {
          atomicReference.set((TdApi.ChatMembers) object).members;
          latch.countDown();
        }
      });
      // handle InterruptedException
      latch.await(10, TimeUnit.SECONDS);
      return atomicReference.get();
    });
    //further steps which need to wait for the result of the step before)
  }

请注意，使用与 相同的技巧CountDownLatch来等待结果。同样，回调的结果应该由锁保护，因为它被不同的线程访问。100% 明确地说，由于附带的原因，这不是必需的CountDownLatch，但是我建议无论如何使用显式同步，例如AtomicReference。