异步调用

某些情况下希望dubbo接口异步调用，避免不必要的等待。

Dubbo 异步调用分为 Provider 端异步调用和 Consumer 端异步两种模式。

  • Consumer 端异步是指发起 RPC 调用后立即返回，调用线程继续处理其他业务逻辑，当响应结果返回后通过回调函数通知消费端结果。
  • Provider 端异步执行将阻塞的业务从 Dubbo 内部线程池切换到业务自定义线程，避免Dubbo线程池的过度占用，有助于避免不同服务间的互相影响。

以下是消费端 consumer 异步调用的工作示例图：

/user-guide/images/future.jpg

Provider 端异步执行和 Consumer 端异步调用是相互独立的，你可以任意正交组合两端配置。

  • Consumer同步 - Provider同步
  • Consumer异步 - Provider同步
  • Consumer同步 - Provider异步
  • Consumer异步 - Provider异步

本文档演示的完整示例源码请参见：

Provider异步

1 使用CompletableFuture

接口定义：

1. public interface AsyncService {
2. /**
3. * 同步调用方法
4. */
5. String invoke(String param);
6. /**
7. * 异步调用方法
8. */
9. CompletableFuture<String> asyncInvoke(String param);
10. }

服务实现：

1. @DubboService
2. public class AsyncServiceImpl implements AsyncService {

4. @Override
5. public String invoke(String param) {
6. try {
7. long time = ThreadLocalRandom.current().nextLong(1000);
8. Thread.sleep(time);
9. StringBuilder s = new StringBuilder();
10. s.append("AsyncService invoke param:").append(param).append(",sleep:").append(time);
11. return s.toString();
12. }
13. catch (InterruptedException e) {
14. Thread.currentThread().interrupt();
15. }
16. return null;
17. }

19. @Override
20. public CompletableFuture<String> asyncInvoke(String param) {
21. // 建议为supplyAsync提供自定义线程池
22. return CompletableFuture.supplyAsync(() -> {
23. try {
24. // Do something
25. long time = ThreadLocalRandom.current().nextLong(1000);
26. Thread.sleep(time);
27. StringBuilder s = new StringBuilder();
28. s.append("AsyncService asyncInvoke param:").append(param).append(",sleep:").append(time);
29. return s.toString();
30. } catch (InterruptedException e) {
31. Thread.currentThread().interrupt();
32. }
33. return null;
34. });
35. }
36. }

通过 return CompletableFuture.supplyAsync() ，业务执行已从 Dubbo 线程切换到业务线程，避免了对 Dubbo 线程池的阻塞。

2 使用AsyncContext

Dubbo 提供了一个类似 Servlet 3.0 的异步接口AsyncContext，在没有 CompletableFuture 签名接口的情况下，也可以实现 Provider 端的异步执行。

接口定义：

1. public interface AsyncService {
2. String sayHello(String name);
3. }

服务实现：

1. public class AsyncServiceImpl implements AsyncService {
2. public String sayHello(String name) {
3. final AsyncContext asyncContext = RpcContext.startAsync();
4. new Thread(() -> {
5. // 如果要使用上下文，则必须要放在第一句执行
6. asyncContext.signalContextSwitch();
7. try {
8. Thread.sleep(500);
9. } catch (InterruptedException e) {
10. e.printStackTrace();
11. }
12. // 写回响应
13. asyncContext.write("Hello " + name + ", response from provider.");
14. }).start();
15. return null;
16. }
17. }

Consumer异步

1 使用CompletableFuture

1. @DubboReference
2. private AsyncService asyncService;

4. @Override
5. public void run(String... args) throws Exception {
6. //调用异步接口
7. CompletableFuture<String> future1 = asyncService.asyncInvoke("async call request1");
8. future1.whenComplete((v, t) -> {
9. if (t != null) {
10. t.printStackTrace();
11. } else {
12. System.out.println("AsyncTask Response-1: " + v);
13. }
14. });
15. //两次调用并非顺序返回
16. CompletableFuture<String> future2 = asyncService.asyncInvoke("async call request2");
17. future2.whenComplete((v, t) -> {
18. if (t != null) {
19. t.printStackTrace();
20. } else {
21. System.out.println("AsyncTask Response-2: " + v);
22. }
23. });
24. //consumer异步调用
25. CompletableFuture<String> future3 =  CompletableFuture.supplyAsync(() -> {
26. return asyncService.invoke("invoke call request3");
27. });
28. future3.whenComplete((v, t) -> {
29. if (t != null) {
30. t.printStackTrace();
31. } else {
32. System.out.println("AsyncTask Response-3: " + v);
33. }
34. });

36. System.out.println("AsyncTask Executed before response return.");
37. }

2 使用 RpcContext

在注解中配置:

1. @DubboReference(async="true")
2. private AsyncService asyncService;

也可以指定方法级别的异步配置：

1. @DubboReference(methods = {@Method(name = "sayHello", timeout = 5000)})
2. private AsyncService asyncService;

接下来的调用即会是异步的：

1. // 此调用会立即返回null
2. asyncService.sayHello("world");
3. // 拿到调用的Future引用，当结果返回后，会被通知和设置到此Future
4. CompletableFuture<String> helloFuture = RpcContext.getServiceContext().getCompletableFuture();
5. // 为Future添加回调
6. helloFuture.whenComplete((retValue, exception) -> {
7. if (exception == null) {
8. System.out.println(retValue);
9. } else {
10. exception.printStackTrace();
11. }
12. });

或者，也可以这样做异步调用

1. CompletableFuture<String> future = RpcContext.getServiceContext().asyncCall(
2. () -> {
3. asyncService.sayHello("oneway call request1");
4. }
5. );

7. future.get();

异步总是不等待返回，你也可以设置是否等待消息发出

  • sent="true" 等待消息发出，消息发送失败将抛出异常。
  • sent="false" 不等待消息发出，将消息放入 IO 队列，即刻返回。
1. @DubboReference(methods = {@Method(name = "sayHello", timeout = 5000， sent = true)})
2. private AsyncService asyncService;

如果你只是想异步，完全忽略返回值，可以配置 return="false"，以减少 Future 对象的创建和管理成本

1. @DubboReference(methods = {@Method(name = "sayHello", timeout = 5000， return = false)})
2. private AsyncService asyncService;

最后修改 September 13, 2024: Refactor website structure (#2860) (1a4b998f54b)