dubbo线程模型分为IO线程和服务调用处理线程，IO线程主要是netty的线程，可以在protocol标签中使用iothreads配置，缺省值为处理器核数+1，服务处理线程主要是提供端netty接收到请求后，处理业务的线程，在dubbo官方文档关于服务调用处理线程的配置如下：
<dubbo:protocol name="dubbo" dispatcher="all" threadpool="fixed" threads="100" />
重要的参数有三个dispatcher、threadpool、threads：
- dispatcher:线程分发器 all(默认)、message、direct(不派发)等
- threadpool:具体处理业务的线程池
- threads:线程池线程数
在消费端其实都是IO线程，并不涉及到服务调用处理线程，在此不做分析，重点分析提供端的线程模型：netty IO线程和服务调用处理线程。
一、netty的IO线程：
@Override
protected void doOpen() throws Throwable {
//boss线程池 负责处理消费端的链接
ExecutorService boss = Executors.newCachedThreadPool(new
NamedThreadFactory("NettyServerBoss", true));
//worker线程池，负责交换数据
ExecutorService worker = Executors.newCachedThreadPool(new
NamedThreadFactory("NettyServerWorker", true));
// 限制worker线程池最大线程数，默认处理器数+1
ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
...........
//绑定端口
channel = bootstrap.bind(getBindAddress());
}
boss线程接受了socket连接求后，会产生一个channel，此channel要与dubbo的区分开来，然后将channel交给ServerSocketChannelFactory后，又继续接受新的请求，ServerSocketChannelFactory则会从worker线程池中找出一个worker线程来继续处理这个请求，worker线程数可以自定义配置。
<dubbo:protocol name="dubbo" iothreads="100" port="20880"/>
二、服务调用处理线程
当消费端调用请求到服务端后，netty接收到请求后初始化链接后，会调用NettyHandler.channelConnected()方法:
@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
//获取一个NettyChannel ctx.getChannel()为boss线程池传递过来的
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
if (channel != null) {//没有直接创建一个 key: ip:port
channels.put(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()), channel);
}
//handler为NettyHandler初始化时传递过来的
handler.connected(channel);
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
static NettyChannel getOrAddChannel(org.jboss.netty.channel.Channel ch, URL url, ChannelHandler handler) {
if (ch == null) {
return null;
}
/* 根据netty中的channel从channelMap中获取dubbo的 NettyChannel
* org.jboss.netty.channel.Channel --> NettyChannel
* see ConcurrentMap<org.jboss.netty.channel.Channel, NettyChannel> channelMap
*/
NettyChannel ret = channelMap.get(ch);
if (ret == null) {//没有直接初始化一个，缓存到channelMap中
NettyChannel nc = new NettyChannel(ch, url, handler);
if (ch.isConnected()) {
ret = channelMap.putIfAbsent(ch, nc);
}
if (ret == null) {
ret = nc;
}
}
return ret;
}
NettyHandler.channelConnected()方法最后会调用handler.connected(channel)-->AbstractServer.connected(channel)：
@Override
public void connected(Channel ch) throws RemotingException {
Collection<Channel> channels = getChannels();
if (accepts > 0 && channels.size() > accepts) {
//服务端建立的长连接超过限制，直接关闭，不再进行下面的处理
logger.error("Close channel " + ch + ", cause: The server " + ch.getLocalAddress() + " connections greater than max config " + accepts);
ch.close();
return;
}
super.connected(ch);
}
上面的方法主要是检查服务提供者接收的长连接数是否超出限制，消费端建立的tcp长连接太多，建立多余的链接会拒绝，消费端会接收到异常，配置的长连接不会像JDK中的线程池那样按需来建立，而是在消费者启动后就全部创建好。消费端参数为connections，服务端为accepts
com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol
private ExchangeClient[] getClients(URL url){
//是否共享连接
boolean service_share_connect = false;
int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);//从url获取connections
//如果connections不配置，则共享连接，否则每服务每连接
if (connections == 0){
service_share_connect = true;
connections = 1;
}
ExchangeClient[] clients = new ExchangeClient[connections];
for (int i = 0; i < clients.length; i++) {//根据connections初始化
if (service_share_connect){
clients[i] = getSharedClient(url);
} else {
clients[i] = initClient(url);
}
}
return clients;
}
链接建立完毕之后，调用NettyHandler.messageReceived()方法：
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
//根据netty的channel取出一个dubbo的NettyChannel
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
handler.received(channel, e.getMessage());//参数分别为NettyChannel RpcInvocation
} finally {//处理完成之后将channel移除
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
handler.received()的handler为nettyServer，初始化的时候传进去的，进而调用到了AbstractPeer.received():
public void received(Channel ch, Object msg) throws RemotingException {
if (closed) {
return;
}
handler.received(ch, msg);
}
上面的handler为初始化 nettyServer时设置的：
public NettyServer(URL url, ChannelHandler handler) throws RemotingException{
super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}
进入到ChannelHandlers.wrap(.. , ..)方法，最后调用到了ChannelHandlers.wrapInternal(.. , ..)，在这个方法里封装的handler层次比较深
protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
.getAdaptiveExtension().dispatch(handler, url)));
}
先看Dispatcher这个接口：
Dispatcher.png
也是通过SPI机制进行适配的，默认实现是AllDispatcher，其他实现对应<dubbo:protocol />中dispatcher参数：
- all 所有消息都派发到线程池，包括请求，响应，连接事件，断开事件，心跳等。
- direct 所有消息都不派发到线程池，全部在 IO 线程上直接执行。
- message 只有请求响应消息派发到线程池，其它连接断开事件，心跳等消息，直接在 IO 线程上执行。
- execution 只请求消息派发到线程池，不含响应，响应和其它连接断开事件，心跳等消息，直接在 IO 线程上执行。
- connection 在 IO 线程上，将连接断开事件放入队列，有序逐个执行，其它消息派发到线程池。
看AllDispatcher默认实现：
/**
* 默认的线程池配置
*
* @author chao.liuc
*/
public class AllDispatcher implements Dispatcher {
public static final String NAME = "all";
public ChannelHandler dispatch(ChannelHandler handler, URL url) {
return new AllChannelHandler(handler, url);
}
}
回到ChannelHandlers.wrapInternal(.. , ..)，最终这个handler的封装层次为：MultiMessageHandler-->HeartbeatHandler-->AllChannelHandler，MultiMessageHandler.received()如果是批量请求，依次调用下一个handler处理，HeartbeatHandler处理心跳检测调用，下面看看AllChannelHandler.received():
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) {
throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
}
}
private ExecutorService getExecutorService() {
ExecutorService cexecutor = executor;//AllChannelHandler初始化的时候设置的
if (cexecutor == null || cexecutor.isShutdown()) {
cexecutor = SHARED_EXECUTOR;
}
return cexecutor;
}
接着看cexecutor的初始化过程，在WrappedChannelHandler的构造方法中：
public WrappedChannelHandler(ChannelHandler handler, URL url) {
.................................................
executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
.................................................
}
image.png
此处也是spi机制，根据url中的threadpool参数，适配出具体的实现：
- CachedThreadPool:此线程池可伸缩，线程空闲一分钟后回收，新请求重新创建线程
- FixedThreadPool: 此线程池启动时即创建固定大小的线程数，不做任何伸缩
- LimitedThreadPool: 此线程池一直增长，直到上限，增长后不收缩
进入默认的FixedThreadPool：
//在Java线程池实现中对应 Executors.newFixedThreadPool()
public class FixedThreadPool implements ThreadPool {
public Executor getExecutor(URL url) {
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
//服务调用线程数
int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
//等待队列里任务数
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
//初始化线程池 池内线程默认200 等待任务使用SynchronousQueue、LinkedBlockingQueue
return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
queues == 0 ? new SynchronousQueue<Runnable>() :
(queues < 0 ? new LinkedBlockingQueue<Runnable>()
: new LinkedBlockingQueue<Runnable>(queues)),
new NamedThreadFactory(name, true), new
//线程池资源枯竭后的处理器
AbortPolicyWithReport(name, url));
}
}
在线程池资源枯竭的时候，会调用AbortPolicyWithReport.rejectedExecution()处理，可以看到打印出了日志，并且抛出了异常，消费端也可以接收到RpcException，因为NettyCodecAdapter.InternalDecoder.exceptionCaught已经对该异常进行了处理，直接输出到了消费者端
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
String msg = String.format("Thread pool is EXHAUSTED!" +
" Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d)," +
" Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!" ,
threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
url.getProtocol(), url.getIp(), url.getPort());
logger.warn(msg);
throw new RejectedExecutionException(msg);//throw exception
}
接下里回到AllChannelHandler.received(..,..)，初始化一个任务，丢到线程池执行，接下来就是调用DecodeHandler.received()->HeaderExchangeHandle.received()-->ExchangeHandlerAdapter.reply()
执行具体的业务。
new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message)
ChannelEventRunnable.run():
public void run() {
.............................
switch (state) {
case RECEIVED:
try{
handler.received(channel, message);
}catch (Exception e) {
logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
+ ", message is "+ message,e);
}
break;
.............................
}
}
参考：
http://blog.sina.com.cn/s/blog_60f11afd01010wtp.html
http://blog.csdn.net/manzhizhen/article/details/73436619
网友评论