消費(fèi)端線程模型，提供者端線程模型

消費(fèi)端線程模型

對(duì) 2.7.5 版本之前的 Dubbo 應(yīng)用，尤其是一些消費(fèi)端應(yīng)用，當(dāng)面臨需要消費(fèi)大量服務(wù)且并發(fā)數(shù)比較大的大流量場(chǎng)景時(shí)（典型如網(wǎng)關(guān)類場(chǎng)景），經(jīng)常會(huì)出現(xiàn)消費(fèi)端線程數(shù)分配過多的問題，具體問題討論可參見?Need a limited Threadpool in consumer side #2013

改進(jìn)后的消費(fèi)端線程池模型，通過復(fù)用業(yè)務(wù)端被阻塞的線程，很好的解決了這個(gè)問題。

老的線程池模型

我們重點(diǎn)關(guān)注 Consumer 部分：

1. 業(yè)務(wù)線程發(fā)出請(qǐng)求，拿到一個(gè) Future 實(shí)例。
2. 業(yè)務(wù)線程緊接著調(diào)用 future.get 阻塞等待業(yè)務(wù)結(jié)果返回。
3. 當(dāng)業(yè)務(wù)數(shù)據(jù)返回后，交由獨(dú)立的 Consumer 端線程池進(jìn)行反序列化等處理，并調(diào)用 future.set 將反序列化后的業(yè)務(wù)結(jié)果置回。
4. 業(yè)務(wù)線程拿到結(jié)果直接返回

當(dāng)前線程池模型

提供端線程模型

Dubbo協(xié)議的和Triple協(xié)議目前的線程模型還并沒有對(duì)齊，下面分開介紹Triple協(xié)議和Dubbo協(xié)議的線程模型。

Dubbo協(xié)議

介紹Dubbo協(xié)議的Provider端線程模型之前，先介紹Dubbo對(duì)channel上的操作抽象成了五種行為：

• 建立連接：connected，主要是的職責(zé)是在channel記錄read、write的時(shí)間，以及處理建立連接后的回調(diào)邏輯，比如dubbo支持在斷開后自定義回調(diào)的hook（onconnect），即在該操作中執(zhí)行。
• 斷開連接：disconnected，主要是的職責(zé)是在channel移除read、write的時(shí)間，以及處理端開連接后的回調(diào)邏輯，比如dubbo支持在斷開后自定義回調(diào)的hook（ondisconnect），即在該操作中執(zhí)行。
• 發(fā)送消息：sent，包括發(fā)送請(qǐng)求和發(fā)送響應(yīng)。記錄write的時(shí)間。
• 接收消息：received，包括接收請(qǐng)求和接收響應(yīng)。記錄read的時(shí)間。
• 異常捕獲：caught，用于處理在channel上發(fā)生的各類異常。

Dubbo框架的線程模型與以上這五種行為息息相關(guān)，Dubbo協(xié)議Provider線程模型可以分為五類，也就是AllDispatcher、DirectDispatcher、MessageOnlyDispatcher、ExecutionDispatcher、ConnectionOrderedDispatcher。

配置方式

線程模型	配置值
All Dispatcher	all
Direct Dispatcher	direct
Execution Dispatcher	execution
Message Only Dispatcher	message
Connection Ordered Dispatcher	connection

拿 application.yaml 的配置方式舉例：在protocol下配置dispatcher: all，即可把dubbo協(xié)議的線程模型調(diào)整為All Dispatcher

dubbo:application:name: dubbo-springboot-demo-providerprotocol:name: dubboport: -1dispatcher: allregistry:id: zk-registryaddress: zookeeper://127.0.0.1:2181

All Dispatcher

下圖是All Dispatcher的線程模型說(shuō)明圖：

?

• 在IO線程中執(zhí)行的操作有：
  1. sent操作在IO線程上執(zhí)行。
  2. 序列化響應(yīng)在IO線程上執(zhí)行。
• 在Dubbo線程中執(zhí)行的操作有：
  1. received、connected、disconnected、caught都是在Dubbo線程上執(zhí)行的。
  2. 反序列化請(qǐng)求的行為在Dubbo中做的。

Direct Dispatcher

下圖是Direct Dispatcher的線程模型說(shuō)明圖：

• 在IO線程中執(zhí)行的操作有：
  1. received、connected、disconnected、caught、sent操作在IO線程上執(zhí)行。
  2. 反序列化請(qǐng)求和序列化響應(yīng)在IO線程上執(zhí)行。
• 1. 并沒有在Dubbo線程操作的行為。

Execution Dispatcher

下圖是Execution Dispatcher的線程模型說(shuō)明圖：

• 在IO線程中執(zhí)行的操作有：
  1. sent、connected、disconnected、caught操作在IO線程上執(zhí)行。
  2. 序列化響應(yīng)在IO線程上執(zhí)行。
• 在Dubbo線程中執(zhí)行的操作有：
  1. received都是在Dubbo線程上執(zhí)行的。
  2. 反序列化請(qǐng)求的行為在Dubbo中做的。

Message Only Dispatcher

在Provider端，Message Only Dispatcher和Execution Dispatcher的線程模型是一致的，所以下圖和Execution Dispatcher的圖一致，區(qū)別在Consumer端。見下方Consumer端的線程模型。

下圖是Message Only Dispatcher的線程模型說(shuō)明圖：

• 在IO線程中執(zhí)行的操作有：
  1. sent、connected、disconnected、caught操作在IO線程上執(zhí)行。
  2. 序列化響應(yīng)在IO線程上執(zhí)行。
• 在Dubbo線程中執(zhí)行的操作有：
  1. received都是在Dubbo線程上執(zhí)行的。
  2. 反序列化請(qǐng)求的行為在Dubbo中做的。

Connection Ordered Dispatcher

下圖是Connection Ordered Dispatcher的線程模型說(shuō)明圖：

?

• 在IO線程中執(zhí)行的操作有：
  1. sent操作在IO線程上執(zhí)行。
  2. 序列化響應(yīng)在IO線程上執(zhí)行。
• 在Dubbo線程中執(zhí)行的操作有：
  1. received、connected、disconnected、caught都是在Dubbo線程上執(zhí)行的。但是connected和disconnected兩個(gè)行為是與其他兩個(gè)行為通過線程池隔離開的。并且在Dubbo connected thread pool中提供了鏈接限制、告警燈能力。
  2. 反序列化請(qǐng)求的行為在Dubbo中做的。

Triple協(xié)議

下圖為Triple協(xié)議 Provider端的線程模型

?

?

Triple協(xié)議Provider線程模型目前還比較簡(jiǎn)單，目前序列化和反序列化操作都在Dubbo線程上工作，而IO線程并沒有承載這些工作。

線程池隔離

一種新的線程池管理方式，使得提供者應(yīng)用內(nèi)各個(gè)服務(wù)的線程池隔離開來(lái)，互相獨(dú)立，某個(gè)服務(wù)的線程池資源耗盡不會(huì)影響其他正常服務(wù)。支持線程池可配置化，由用戶手動(dòng)指定。

使用線程池隔離來(lái)確保 Dubbo 用于調(diào)用遠(yuǎn)程方法的線程與微服務(wù)用于執(zhí)行其任務(wù)的線程是分開的?？梢酝ㄟ^防止線程阻塞或相互競(jìng)爭(zhēng)來(lái)幫助提高系統(tǒng)的性能和穩(wěn)定性。

目前可以以 API、XML、Annotation 的方式進(jìn)行配置

配置參數(shù)

• ApplicationConfig?新增?String executor-management-mode?參數(shù)，配置值為?default?和?isolation?，默認(rèn)為?default。
  • executor-management-mode = default?使用原有?以協(xié)議端口為粒度、服務(wù)間共享?的線程池管理方式
  • executor-management-mode = isolation?使用新增的?以服務(wù)三元組為粒度、服務(wù)間隔離?的線程池管理方式
• ServiceConfig?新增?Executor executor?參數(shù)，用以服務(wù)間隔離的線程池，可以由用戶配置化、提供自己想要的線程池，若沒有指定，則會(huì)根據(jù)協(xié)議配置(ProtocolConfig)信息構(gòu)建默認(rèn)的線程池用以服務(wù)隔離。

ServiceConfig?新增?Executor executor?配置參數(shù)只有指定executor-management-mode = isolation?才生效。

API

public void test() {// provider appDubboBootstrap providerBootstrap = DubboBootstrap.newInstance();ServiceConfig serviceConfig1 = new ServiceConfig();serviceConfig1.setInterface(DemoService.class);serviceConfig1.setRef(new DemoServiceImpl());serviceConfig1.setVersion(version1);// set executor1 for serviceConfig1, max threads is 10NamedThreadFactory threadFactory1 = new NamedThreadFactory("DemoService-executor");ExecutorService executor1 = Executors.newFixedThreadPool(10, threadFactory1);serviceConfig1.setExecutor(executor1);ServiceConfig serviceConfig2 = new ServiceConfig();serviceConfig2.setInterface(HelloService.class);serviceConfig2.setRef(new HelloServiceImpl());serviceConfig2.setVersion(version2);// set executor2 for serviceConfig2, max threads is 100NamedThreadFactory threadFactory2 = new NamedThreadFactory("HelloService-executor");ExecutorService executor2 = Executors.newFixedThreadPool(100, threadFactory2);serviceConfig2.setExecutor(executor2);ServiceConfig serviceConfig3 = new ServiceConfig();serviceConfig3.setInterface(HelloService.class);serviceConfig3.setRef(new HelloServiceImpl());serviceConfig3.setVersion(version3);// Because executor is not set for serviceConfig3, the default executor of serviceConfig3 is built using// the threadpool parameter of the protocolConfig ( FixedThreadpool , max threads is 200)serviceConfig3.setExecutor(null);// It takes effect only if [executor-management-mode=isolation] is configuredApplicationConfig applicationConfig = new ApplicationConfig("provider-app");applicationConfig.setExecutorManagementMode("isolation");providerBootstrap.application(applicationConfig).registry(registryConfig)// export with tri and dubbo protocol.protocol(new ProtocolConfig("tri", 20001)).protocol(new ProtocolConfig("dubbo", 20002)).service(serviceConfig1).service(serviceConfig2).service(serviceConfig3);providerBootstrap.start();}

?

XML

<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"xmlns="http://www.springframework.org/schema/beans"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsdhttp://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd"><!-- NOTE: we need config executor-management-mode="isolation" --><dubbo:application name="demo-provider" executor-management-mode="isolation"></dubbo:application><dubbo:config-center address="zookeeper://127.0.0.1:2181"/><dubbo:metadata-report address="zookeeper://127.0.0.1:2181"/><dubbo:registry id="registry1" address="zookeeper://127.0.0.1:2181?registry-type=service"/><dubbo:protocol name="dubbo" port="-1"/><dubbo:protocol name="tri" port="-1"/><!-- expose three service with dubbo and tri protocol--><bean id="demoServiceV1" class="org.apache.dubbo.config.spring.impl.DemoServiceImpl"/><bean id="helloServiceV2" class="org.apache.dubbo.config.spring.impl.HelloServiceImpl"/><bean id="helloServiceV3" class="org.apache.dubbo.config.spring.impl.HelloServiceImpl"/><!-- customized thread pool --><bean id="executor-demo-service"class="org.apache.dubbo.config.spring.isolation.spring.support.DemoServiceExecutor"/><bean id="executor-hello-service"class="org.apache.dubbo.config.spring.isolation.spring.support.HelloServiceExecutor"/><!-- this service use [executor="executor-demo-service"] as isolated thread pool--><dubbo:service executor="executor-demo-service"interface="org.apache.dubbo.config.spring.api.DemoService" version="1.0.0" group="Group1"timeout="3000" ref="demoServiceV1" registry="registry1" protocol="dubbo,tri"/><!-- this service use [executor="executor-hello-service"] as isolated thread pool--><dubbo:service executor="executor-hello-service"interface="org.apache.dubbo.config.spring.api.HelloService" version="2.0.0" group="Group2"timeout="5000" ref="helloServiceV2" registry="registry1" protocol="dubbo,tri"/><!-- not set executor for this service, the default executor built using threadpool parameter of the protocolConfig --><dubbo:service interface="org.apache.dubbo.config.spring.api.HelloService" version="3.0.0" group="Group3"timeout="5000" ref="helloServiceV3" registry="registry1" protocol="dubbo,tri"/></beans>

Annotation

@Configuration
@EnableDubbo(scanBasePackages = "org.apache.dubbo.config.spring.isolation.spring.annotation.provider")
public class ProviderConfiguration {@Beanpublic RegistryConfig registryConfig() {RegistryConfig registryConfig = new RegistryConfig();registryConfig.setAddress("zookeeper://127.0.0.1:2181");return registryConfig;}// NOTE: we need config executor-management-mode="isolation"@Beanpublic ApplicationConfig applicationConfig() {ApplicationConfig applicationConfig = new ApplicationConfig("provider-app");applicationConfig.setExecutorManagementMode("isolation");return applicationConfig;}// expose services with dubbo protocol@Beanpublic ProtocolConfig dubbo() {ProtocolConfig protocolConfig = new ProtocolConfig("dubbo");return protocolConfig;}// expose services with tri protocol@Beanpublic ProtocolConfig tri() {ProtocolConfig protocolConfig = new ProtocolConfig("tri");return protocolConfig;}// customized thread pool@Bean("executor-demo-service")public Executor demoServiceExecutor() {return new DemoServiceExecutor();}// customized thread pool@Bean("executor-hello-service")public Executor helloServiceExecutor() {return new HelloServiceExecutor();}
}

// customized thread pool
public class DemoServiceExecutor extends ThreadPoolExecutor {public DemoServiceExecutor() {super(10, 10, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(),new NamedThreadFactory("DemoServiceExecutor"));}
}

// customized thread pool
public class HelloServiceExecutor extends ThreadPoolExecutor {public HelloServiceExecutor() {super(100, 100, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(),new NamedThreadFactory("HelloServiceExecutor"));}
}

// "executor-hello-service" is beanName
@DubboService(executor = "executor-demo-service", version = "1.0.0", group = "Group1")
public class DemoServiceImplV1 implements DemoService {@Overridepublic String sayName(String name) {return "server name";}@Overridepublic Box getBox() {return null;}
}

// not set executor for this service, the default executor built using threadpool parameter of the protocolConfig
@DubboService(version = "3.0.0", group = "Group3")
public class HelloServiceImplV2 implements HelloService {private static final Logger logger = LoggerFactory.getLogger(HelloServiceImplV2.class);@Overridepublic String sayHello(String name) {return "server hello";}
}

@DubboService(executor = "executor-hello-service", version = "2.0.0", group = "Group2")
public class HelloServiceImplV3 implements HelloService {private static final Logger logger = LoggerFactory.getLogger(HelloServiceImplV3.class);@Overridepublic String sayHello(String name) {return "server hello";}
}

線程池狀態(tài)導(dǎo)出

dubbo 通過 Jstack 自動(dòng)導(dǎo)出線程堆棧來(lái)保留現(xiàn)場(chǎng)，方便排查問題。

默認(rèn)策略

• 導(dǎo)出路徑: user.home標(biāo)識(shí)的用戶主目錄
• 導(dǎo)出間隔: 最短間隔允許每隔10分鐘導(dǎo)出一次
• 導(dǎo)出開關(guān): 默認(rèn)打開

當(dāng)業(yè)務(wù)線程池滿時(shí)，我們需要知道線程都在等待哪些資源、條件，以找到系統(tǒng)的瓶頸點(diǎn)或異常點(diǎn)。

導(dǎo)出開關(guān)控制

# dubbo.properties
dubbo.application.dump.enable=true

<dubbo:application name="demo-provider" dump-enable="false"/>

dubbo:application:name: dubbo-springboot-demo-providerdump-enable: false

導(dǎo)出路徑

# dubbo.properties
dubbo.application.dump.directory=/tmp

<dubbo:application name="demo-provider" dump-directory="/tmp"/>

dubbo:application:name: dubbo-springboot-demo-providerdump-directory: /tmp