DUBBO的多种线程模型
### 背景
DUBBO提供了多种线程模型，选择线程模型需要在配置文件指定dispatcher属性：
```java
<dubbo:protocol name="dubbo" dispatcher="all" />
<dubbo:protocol name="dubbo" dispatcher="direct" />
<dubbo:protocol name="dubbo" dispatcher="message" />
<dubbo:protocol name="dubbo" dispatcher="execution" />
<dubbo:protocol name="dubbo" dispatcher="connection" />
```
不同线程模型在选择是使用IO线程还是业务线程，DUBBO官网文档说明：
```java
all
所有消息都派发到业务线程池，包括请求，响应，连接事件，断开事件，心跳
direct
所有消息都不派发到业务线程池，全部在IO线程直接执行
message
只有请求响应消息派发到业务线程池，其它连接断开事件，心跳等消息直接在IO线程执行
execution
只有请求消息派发到业务线程池，响应和其它连接断开事件，心跳等消息直接在IO线程执行
connection
在IO线程上将连接断开事件放入队列，有序逐个执行，其它消息派发到业务线程池
```
生产者和消费者默认线程模型都会使用All Dispatcher
### 确定时机
生产者和消费者在初始化时确定线程模型：
```java
// 生产者
public class NettyServer extends AbstractServer implements Server {
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}
}
// 消费者
public class NettyClient extends AbstractClient {
public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException {
super(url, wrapChannelHandler(url, handler));
}
}
```
生产者和消费者默认线程模型都会使用AllDispatcher，ChannelHandlers.wrap方法可以获取Dispatch自适应扩展点。如果我们在配置文件中指定dispatcher，扩展点加载器会从URL获取属性值加载对应线程模型。本文以生产者为例进行分析：
```
java
public class NettyServer extends AbstractServer implements Server {
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
// ChannelHandlers.wrap确定线程策略
super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}
}
public class ChannelHandlers {
protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class).getAdaptiveExtension().dispatch(handler, url)));
}
}
@SPI(AllDispatcher.NAME)
public interface Dispatcher {
@Adaptive({Constants.DISPATCHER_KEY, "channel.handler"})
ChannelHandler dispatch(ChannelHandler handler, URL url);
}
```
### 源码分析
我们分析其中两个线程模型源码，其它线程模型请阅读DUBBO源码。AllDispatcher模型所有消息都派发到业务线程池，包括请求，响应，连接事件，断开事件，心跳：
```java
public class AllDispatcher implements Dispatcher {
// 线程模型名称
public static final String NAME = "all";
// 具体实现策略
@Override
public ChannelHandler dispatch(ChannelHandler handler, URL url) {
return new AllChannelHandler(handler, url);
}
}
public class AllChannelHandler extends WrappedChannelHandler {
@Override
public void connected(Channel channel) throws RemotingException {
// 连接完成事件交给业务线程池
ExecutorService cexecutor = getExecutorService();
try {
cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED));
} catch (Throwable t) {
throw new ExecutionException("connect event", channel, getClass() + " error when process connected event", t);
}
}
@Override
public void disconnected(Channel channel) throws RemotingException {
// 断开连接事件交给业务线程池
ExecutorService cexecutor = getExecutorService();
try {
cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.DISCONNECTED));
} catch (Throwable t) {
throw new ExecutionException("disconnect event", channel, getClass() + " error when process disconnected event", t);
}
}
@Override
public void received(Channel channel, Object message) throws RemotingException {
// 请求响应事件交给业务线程池
ExecutorService cexecutor = getExecutorService();
try {
cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
} catch (Throwable t) {
if(message instanceof Request && t instanceof RejectedExecutionException) {
Request request = (Request)message;
if(request.isTwoWay()) {
String msg = "Server side(" + url.getIp() + "," + url.getPort() + ") threadpool is exhausted ,detail msg:" + t.getMessage();
Response response = new Response(request.getId(), request.getVersion());
response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR);
response.setErrorMessage(msg);
channel.send(response);
return;
}
}
throw new ExecutionException(message, channel, getClass() + " error when process received event", t);
}
}
@Override
public void caught(Channel channel, Throwable exception) throws RemotingException {
// 异常事件交给业务线程池
ExecutorService cexecutor = getExecutorService();
try {
cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CAUGHT, exception));
} catch (Throwable t) {
throw new ExecutionException("caught event", channel, getClass() + " error when process caught event", t);
}
}
}
```
DirectDispatcher策略所有消息都不派发到业务线程池，全部在IO线程直接执行：
```java
public class DirectDispatcher implements Dispatcher {
// 线程模型名称
public static final String NAME = "direct";
// 具体实现策略
@Override
public ChannelHandler dispatch(ChannelHandler handler, URL url) {
// 直接返回handler表示所有事件都交给IO线程处理
return handler;
}
}
```
公众号进入网站体验更佳
