在 Java 中同步多个异步请求

Posted 2023-02-19

【中文标题】在 Java 中同步多个异步请求

【英文标题】：Syncing multiple asynchronous requests in Java

【发布时间】：2019-01-05 12:37:34

【问题描述】：

我正在使用 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 中的任何同步请求都很陌生，我想知道以下几点：

调用此方法并等待收到所有结果后再继续操作的合适方法是什么？

通常，当连续调用多个请求并在继续下一个请求之前等待每个结果时，通常的方法是什么而不是将请求嵌套在彼此内部以在 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
                    
                );
            
);

谢谢！

【问题讨论】：

Client.ResultHandler是接口吗？

是的。你可以在这里看到它：github.com/tdlib/td/blob/master/example/java/org/drinkless/…

【参考方案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。