Dubbo3 異步執(zhí)行

Dubbo 服務(wù)提供方的異步執(zhí)行

Provider端異步執(zhí)行將阻塞的業(yè)務(wù)從Dubbo內(nèi)部線程池切換到業(yè)務(wù)自定義線程，避免Dubbo線程池的過度占用，有助于避免不同服務(wù)間的互相影響。異步執(zhí)行無(wú)異于節(jié)省資源或提升RPC響應(yīng)性能，因?yàn)槿绻麡I(yè)務(wù)執(zhí)行需要阻塞，則始終還是要有線程來負(fù)責(zé)執(zhí)行。

注意
Provider 端異步執(zhí)行和 Consumer 端異步調(diào)用是相互獨(dú)立的，你可以任意正交組合兩端配置

• Consumer同步 - Provider同步

• Consumer異步 - Provider同步

• Consumer同步 - Provider異步

• Consumer異步 - Provider異步

定義 CompletableFuture 簽名的接口

服務(wù)接口定義：

public interface AsyncService {
    CompletableFuture<String> sayHello(String name);
}

服務(wù)實(shí)現(xiàn)：

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()? ，業(yè)務(wù)執(zhí)行已從 Dubbo 線程切換到業(yè)務(wù)線程，避免了對(duì) Dubbo 線程池的阻塞。

使用AsyncContext

Dubbo 提供了一個(gè)類似 Servlet 3.0 的異步接口?AsyncContext?，在沒有?? CompletableFuture?? 簽名接口的情況下，也可以實(shí)現(xiàn) Provider 端的異步執(zhí)行。

服務(wù)接口定義：

public interface AsyncService {
    String sayHello(String name);
}

服務(wù)暴露，和普通服務(wù)完全一致：

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

服務(wù)實(shí)現(xiàn)：

public class AsyncServiceImpl implements AsyncService {
    public String sayHello(String name) {
        final AsyncContext asyncContext = RpcContext.startAsync();
        new Thread(() -> {
            // 如果要使用上下文，則必須要放在第一句執(zhí)行
            asyncContext.signalContextSwitch();
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 寫回響應(yīng)
            asyncContext.write("Hello " + name + ", response from provider.");
        }).start();
        return null;
    }
}