开篇
这篇文章的主要目的是记录下Dubbo当中NettyServer针对请求的处理过程，主要是理顺连接建立的过程和调用处理的过程。
这个解析的核心在于理解类的继承关系（可以理解为设计模式当中的装饰者模式），层层递进理解调用关系。另外通过时序图能够帮助理解整个时序图的调用过程。
NettyServer
NettyServer类依赖图
public abstract class AbstractPeer implements Endpoint, ChannelHandler {
private final ChannelHandler handler;
public AbstractPeer(URL url, ChannelHandler handler) {
this.handler = handler;
}
public void connected(Channel ch) throws RemotingException {
if (closed) {
return;
}
handler.connected(ch);
}
public void disconnected(Channel ch) throws RemotingException {
handler.disconnected(ch);
}
public void sent(Channel ch, Object msg) throws RemotingException {
if (closed) {
return;
}
handler.sent(ch, msg);
}
public void received(Channel ch, Object msg) throws RemotingException {
if (closed) {
return;
}
handler.received(ch, msg);
}
public void caught(Channel ch, Throwable ex) throws RemotingException {
handler.caught(ch, ex);
}
}
说明：
- Dubbo中NettyServer的继承关系图如上图所示，核心的变量在基类当中。
- 关注最顶层的基类AbstractPeer变量包含ChannelHandler对象。
- AbstractPeer的ChannelHandler变量是由NettyServer的构造函数中传入的。
public class NettyServer extends AbstractServer implements Server {
private Map<String, Channel> channels; // <ip:port, channel>
private ServerBootstrap bootstrap;
private org.jboss.netty.channel.Channel channel;
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}
@Override
protected void doOpen() throws Throwable {
NettyHelper.setNettyLoggerFactory();
ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
bootstrap = new ServerBootstrap(channelFactory);
// NettyHandler传入的是NettyServer对象
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
channels = nettyHandler.getChannels();
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
// 核心关键nettyHandler
pipeline.addLast("handler", nettyHandler);
return pipeline;
}
});
// bind
channel = bootstrap.bind(getBindAddress());
}
}
public class ChannelHandlers {
private static ChannelHandlers INSTANCE = new ChannelHandlers();
public static ChannelHandler wrap(ChannelHandler handler, URL url) {
return ChannelHandlers.getInstance().wrapInternal(handler, url);
}
protected static ChannelHandlers getInstance() {
return INSTANCE;
}
protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
.getAdaptiveExtension().dispatch(handler, url)));
}
}
public class AllDispatcher implements Dispatcher {
public static final String NAME = "all";
public ChannelHandler dispatch(ChannelHandler handler, URL url) {
return new AllChannelHandler(handler, url);
}
}
- 核心需要关注NettyServer的构造函数和doOpen()方法。
- NettyServer的构造函数传入的是DecodeHandler对象。
- NettyServer的构造函数通过ChannelHandlers.wrap()封装ChannelHandler对象。
- wrapInternal()方法按照MultiMessageHandler包含HeartbeatHandler，HeartbeatHandler包含AllChannelHandler的顺序进行封装，类似于装饰者模式。
- getAdaptiveExtension().dispatch()方法返回AllChannelHandler对象。
- NettyHandler的构造方法将NettyServer的this作为ChannelHandler对象传参。
- pipeline.addLast("handler", nettyHandler)将nettyHandler加到Netty的职责链中。
- NettyHandler作为处理的入口函数，NettyServer的this自身包括处理的逻辑。
NettyTransporter
public class HeaderExchanger implements Exchanger {
public static final String NAME = "header";
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
}
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}
}
public class NettyTransporter implements Transporter {
public static final String NAME = "netty";
public Server bind(URL url, ChannelHandler listener) throws RemotingException {
return new NettyServer(url, listener);
}
public Client connect(URL url, ChannelHandler listener) throws RemotingException {
return new NettyClient(url, listener);
}
}
说明：
- NettyTransporter的目的是为了打通和NettyServer的初始化过程形成串联。
- NettyTransporter的bind()方法传入的是DecodeHandler对象。
- DecodeHandler的构造函数参数是HeaderExchangeHandler对象。
- HeaderExchangeHandler的构造函数参数是DubboProtocol对象。
- NettyServer的构造函数的参数是DecodeHandler对象。
NettyHandler
public class NettyHandler extends SimpleChannelHandler {
private final Map<String, Channel> channels = new ConcurrentHashMap<String, Channel>(); // <ip:port, channel>
private final URL url;
private final ChannelHandler handler;
public NettyHandler(URL url, ChannelHandler handler) {
this.url = url;
this.handler = handler;
}
@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
if (channel != null) {
channels.put(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()), channel);
}
handler.connected(channel);
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
@Override
public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
channels.remove(NetUtils.toAddressString((InetSocketAddress) ctx.getChannel().getRemoteAddress()));
handler.disconnected(channel);
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
handler.received(channel, e.getMessage());
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
@Override
public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
super.writeRequested(ctx, e);
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
handler.sent(channel, e.getMessage());
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
handler.caught(channel, e.getCause());
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
}
说明
- NettyHandler的channelConnected和channelDisconnected用于处理连接。
- NettyHandler的messageReceived和writeRequested用于处理消息。
- NettyHandler的exceptionCaught用于处理异常。
- NettyHandler构造函数的的参数handler是NettyServer对象。
- NettyHandler的操作通过调用NettyServer的方法connected/received完成。
Dubbo-Netty的Receive过程
Receive过程
说明：
- 1.NettyHandler
- 2.NettyServer
- 3.MultiMessageHandler
- 4.HeartbeatHandler
- 5.AllChannelHandler
- 6.ChannelEventRunnable
- 7.DecodeHandler
- 8.HeaderExchangeHandler
- 9.DubboProtocol.ExchangeHandlerAdapter
Dubbo-Netty的Connect过程
Connect过程.jpg
说明：
- 1.NettyHandler
- 2.NettyServer
- 3.MultiMessageHandler
- 4.HeartbeatHandler
- 5.AllChannelHandler
- 6.ChannelEventRunnable
- 7.DecodeHandler
- 8.HeaderExchangeHandler
- 9.DubboProtocol.ExchangeHandlerAdapter
AllChannelHandler
public class AllChannelHandler extends WrappedChannelHandler {
public AllChannelHandler(ChannelHandler handler, URL url) {
super(handler, url);
}
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);
}
}
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);
}
}
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) {
//TODO A temporary solution to the problem that the exception information can not be sent to the opposite end after the thread pool is full. Need a refactoring
//fix The thread pool is full, refuses to call, does not return, and causes the consumer to wait for time out
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);
}
}
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);
}
}
private ExecutorService getExecutorService() {
ExecutorService cexecutor = executor;
if (cexecutor == null || cexecutor.isShutdown()) {
cexecutor = SHARED_EXECUTOR;
}
return cexecutor;
}
}
说明：
- AllChannelHandler内部通过线程池提交任务的方法进行处理。
- 提交线程池的任务统一封装成ChannelEventRunnable对象。
ChannelEventRunnable
public class ChannelEventRunnable implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(ChannelEventRunnable.class);
private final ChannelHandler handler;
private final Channel channel;
private final ChannelState state;
private final Throwable exception;
private final Object message;
public ChannelEventRunnable(Channel channel, ChannelHandler handler, ChannelState state) {
this(channel, handler, state, null);
}
public ChannelEventRunnable(Channel channel, ChannelHandler handler, ChannelState state, Object message) {
this(channel, handler, state, message, null);
}
public ChannelEventRunnable(Channel channel, ChannelHandler handler, ChannelState state, Throwable t) {
this(channel, handler, state, null, t);
}
public ChannelEventRunnable(Channel channel, ChannelHandler handler, ChannelState state, Object message, Throwable exception) {
this.channel = channel;
this.handler = handler;
this.state = state;
this.message = message;
this.exception = exception;
}
public void run() {
switch (state) {
case CONNECTED:
try {
handler.connected(channel);
} catch (Exception e) {
logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e);
}
break;
case DISCONNECTED:
try {
handler.disconnected(channel);
} catch (Exception e) {
logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e);
}
break;
case SENT:
try {
handler.sent(channel, message);
} catch (Exception e) {
logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
+ ", message is " + message, e);
}
break;
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;
case CAUGHT:
try {
handler.caught(channel, exception);
} catch (Exception e) {
logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
+ ", message is: " + message + ", exception is " + exception, e);
}
break;
default:
logger.warn("unknown state: " + state + ", message is " + message);
}
}
/**
* ChannelState
*
*
*/
public enum ChannelState {
/**
* CONNECTED
*/
CONNECTED,
/**
* DISCONNECTED
*/
DISCONNECTED,
/**
* SENT
*/
SENT,
/**
* RECEIVED
*/
RECEIVED,
/**
* CAUGHT
*/
CAUGHT
}
}
说明：
- ChannelEventRunnable统计封装所有处理事件。
