在 Java 中同步多个异步请求

Question

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

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，好像就是你要找的。​​p>

  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。