开篇
这篇文章的目的是尝试讲解Dubbo Protocol使用Netty Server作为服务端的服务发布过程，为了便于说明清楚，在网上找了一张Dubbo服务发布的流程图，这里我们关注Protocol到Server的部分逻辑，其他的分布其他文章。
DubboProtocol
public class DubboProtocol extends AbstractProtocol {
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
URL url = invoker.getUrl();
// 生成exporter对象并保存至exporterMap
String key = serviceKey(url);
DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
exporterMap.put(key, exporter);
// 开始开启服务端监控
openServer(url);
optimizeSerialization(url);
return exporter;
}
private void openServer(URL url) {
// find server.
String key = url.getAddress();
//client can export a service which's only for server to invoke
boolean isServer = url.getParameter(IS_SERVER_KEY, true);
if (isServer) {
ExchangeServer server = serverMap.get(key);
if (server == null) {
synchronized (this) {
server = serverMap.get(key);
if (server == null) {
// 通过createServer创建Server对象
serverMap.put(key, createServer(url));
}
}
} else {
// server supports reset, use together with override
server.reset(url);
}
}
}
private ExchangeServer createServer(URL url) {
// 省略非核心代码
ExchangeServer server;
try {
// 服务发布的入口，requestHandler包含invokerMap的对象
// server 是HeaderExchangeServer对象
server = Exchangers.bind(url, requestHandler);
} catch (RemotingException e) {
throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
}
return server;
}
private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
// 省略代码
}
}- DubboProtocol的export()方法生成DubboExporter对象并openServer()开启Server。
- openServer()方法内部执行createServer(url)创建Server对象。
- createServer()方法内部生成ExchangeServer对象。
- ExchangeServer对象通过Exchangers.bind(url, requestHandler)生成。
- requestHandler是ExchangeHandlerAdapter对象，内部包含invokerMap对象。
- server对象在这里我们可以看出来是ExchangeServer，通过Exchangers生成。
Exchangers
public class Exchangers {
public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handler == null) {
throw new IllegalArgumentException("handler == null");
}
url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
return getExchanger(url).bind(url, handler);
}
// String DEFAULT_EXCHANGER = "header";
public static Exchanger getExchanger(URL url) {
String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER);
return getExchanger(type);
}
// header=org.apache.dubbo.remoting.exchange.support.header.HeaderExchanger
public static Exchanger getExchanger(String type) {
return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);
}
}- Exchangers的bind()方法内部执行getExchanger(url).bind(url, handler)。
- getExchanger(url).bind(url, handler)中getExchange()通过ExtensionLoader获取Exchanger扩展，这里是HeaderExchanger对象。
- 继续执行HeaderExchanger的bind()方法。
- 这里的bind()方法的参数是ExchangeHandlerAdapter的requestHandler。
public class HeaderExchanger implements Exchanger {
public static final String NAME = "header";
@Override
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeClient(Transporters.connect(url,
new DecodeHandler(new HeaderExchangeHandler(handler))), true);
}
@Override
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeServer(Transporters.bind(url,
new DecodeHandler(new HeaderExchangeHandler(handler))));
}
}- 继续执行HeaderExchanger的bind()方法，生成HeaderExchangeServer对象。
- HeaderExchangeServer对象的参数通过Transporters.bind()生成的Server对象。
- bind()的参数是通过DecodeHandler => HeaderExchangeHandler => requestHandler包装生成的DecodeHandler对象。
- 继续观察Transporters.bind()方法。
Transporters
public class Transporters {
public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handlers == null || handlers.length == 0) {
throw new IllegalArgumentException("handlers == null");
}
ChannelHandler handler;
if (handlers.length == 1) {
handler = handlers[0];
} else {
handler = new ChannelHandlerDispatcher(handlers);
}
return getTransporter().bind(url, handler);
}
public static Transporter getTransporter() {
return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
}
}- Transporters.bind()方法内部执行getTransporter().bind(url, handler)。
- 调用通过getTransporter()方法返回Transporter$Adaptive。
- getTransporter().bind()执行Transporter$Adaptive的bind()方法。
- 继续观察Transporter$Adaptive的bind()方法
public class Transporter$Adaptive implements Transporter {
public Server bind(URL uRL, ChannelHandler channelHandler)
throws RemotingException {
if (uRL == null) {
throw new IllegalArgumentException("url == null");
}
URL uRL2 = uRL;
String string = uRL2.getParameter("server",
uRL2.getParameter("transporter", "netty"));
Transporter transporter = (Transporter) ExtensionLoader
.getExtensionLoader(Transporter.class)
.getExtension(string);
return transporter.bind(uRL, channelHandler);
}
}
public class NettyTransporter implements Transporter {
public static final String NAME = "netty3";
@Override
public Server bind(URL url, ChannelHandler listener) throws RemotingException {
return new NettyServer(url, listener);
}
}- Transporter$Adaptive的bind()方法内部执行获取Transporter扩展并执行bind()方法。
- ExtensionLoader.getExtensionLoader().getExtension()返回NettyTransporter对象。
- 这里的ChannelHandler为DecodeHandler对象。
- 继续执行NettyTransporter的bind()方法，内部创建NettyServer对象。
NettyServer
public class NettyServer extends AbstractServer implements Server {
private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
// 保存连接的Channel信息
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(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
bootstrap = new ServerBootstrap(channelFactory);
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
channels = nettyHandler.getChannels();
// https://issues.jboss.org/browse/NETTY-365
// https://issues.jboss.org/browse/NETTY-379
// final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
bootstrap.setOption("child.tcpNoDelay", true);
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
ChannelPipeline pipeline = Channels.pipeline();
/*int idleTimeout = getIdleTimeout();
if (idleTimeout > 10000) {
pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0));
}*/
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
pipeline.addLast("handler", nettyHandler);
return pipeline;
}
});
// bind
channel = bootstrap.bind(getBindAddress());
}- 创建NettyServer对象并通过doOpen()方法创建Netty Server对象。
- doOpen()方法内部创建ServerBootstrap对象并实现bind启动服务监听。
- NettyServer构造函数内部通过ChannelHandlers.wrap()包装ChannelHandler。
public class ChannelHandlers {
// 关注核心代码逻辑
protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
.getAdaptiveExtension().dispatch(handler, url)));
}
}
public class Dispatcher$Adaptive implements Dispatcher {
public ChannelHandler dispatch(ChannelHandler channelHandler, URL uRL) {
if (uRL == null) {
throw new IllegalArgumentException("url == null");
}
URL uRL2 = uRL;
String string = uRL2.getParameter("dispatcher",
uRL2.getParameter("dispather",
uRL2.getParameter("channel.handler", "all")));
Dispatcher dispatcher = (Dispatcher) ExtensionLoader.getExtensionLoader(Dispatcher.class)
.getExtension(string);
return dispatcher.dispatch(channelHandler, uRL);
}
}
public class AllDispatcher implements Dispatcher {
public static final String NAME = "all";
@Override
public ChannelHandler dispatch(ChannelHandler handler, URL url) {
return new AllChannelHandler(handler, url);
}
}- wrapInternal()的handler参数通过MultiMessageHandler => HeartbeatHandler => AllChannelHandler => DecodeHandler层层封装。
- Dispatcher$Adaptive的ExtensionLoader.getExtensionLoader().getExtension()返回AllDispatcher对象。
- AllDispatcher的dispatch()方法返回AllChannelHandler对象。
- AllChannelHandler封装DecodeHandler对象。
public abstract class AbstractServer extends AbstractEndpoint implements Server {
protected static final String SERVER_THREAD_POOL_NAME = "DubboServerHandler";
private static final Logger logger = LoggerFactory.getLogger(AbstractServer.class);
ExecutorService executor;
private InetSocketAddress localAddress;
private InetSocketAddress bindAddress;
// 最大连接数
private int accepts;
private int idleTimeout;
public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
super(url, handler);
localAddress = getUrl().toInetSocketAddress();
String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost());
int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort());
if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
bindIp = ANYHOST_VALUE;
}
bindAddress = new InetSocketAddress(bindIp, bindPort);
// 最大连接数从URL当中获取
this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS);
this.idleTimeout = url.getParameter(IDLE_TIMEOUT_KEY, DEFAULT_IDLE_TIMEOUT);
try {
// 开启服务监听
doOpen();
} catch (Throwable t) {
}
//fixme replace this with better method
DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
executor = (ExecutorService) dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY, Integer.toString(url.getPort()));
}
}- NettyServer的父类AbstractServer包含了核心的最大连接数的参数。
- AbstractServer的调用抽象方法doOpen()调用子类NettyServer启动服务监听。
handler封装关系图
- 描述创建Server服务 过程中Handler的封装关系。
