dubbo2.7 43 异步执行

有的时候博客内容会有变动，首发博客是最新的，其他博客地址可能会未同步, 认准https://blog.zysicyj.top

全网最细面试题手册，支持艾宾浩斯记忆法。这是一份最全面、最详细、最高质量的 java 面试题，不建议你死记硬背，只要每天复习一遍，有个大概印象就行了。 https://store.amazingmemo.com/chapterDetail/1685324709017001`

Dubbo 服务提供方的异步执行

背景

Provider 端异步执行将阻塞的业务从 Dubbo 内部线程池切换到业务自定义线程，避免 Dubbo 线程池的过度占用，有助于避免不同服务间的互相影响。异步执行无异于节省资源或提升 RPC 响应性能，因为如果业务执行需要阻塞，则始终还是要有线程来负责执行。

注意

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

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

示例

定义 CompletableFuture 签名的接口

服务接口定义：

1
2
3
4
public interface AsyncService {
    CompletableFuture<String> sayHello(String name);
}

服务实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public class AsyncServiceImpl implements AsyncService {
    @Override
    public CompletableFuture<String> sayHello(String name) {
        RpcContext savedContext = RpcContext.getContext();
        // 建议为 supplyAsync 提供自定义线程池，避免使用 JDK 公用线程池
        return CompletableFuture.supplyAsync(() -> {
            System.out.println(savedContext.getAttachment("consumer-key1"));
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "async response from provider.";
        });
    }
}

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

使用 AsyncContext

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

服务接口定义：

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

服务暴露，和普通服务完全一致：

1
2
3
<bean id="asyncService" class="org.apache.dubbo.samples.governance.impl.AsyncServiceImpl"/>
<dubbo:service interface="org.apache.dubbo.samples.governance.api.AsyncService" ref="asyncService"/>

服务实现：

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