前言
Dubbo源码阅读分享系列文章，欢迎大家关注点赞
SPI实现部分
注册中心
通信
RPC
AbstractProtocol
在介绍RPC核心接口的时候我们说过Protocol核心作用是将Invoker服务暴露出去以及引用服务将Invoker对象返回，因此我们就从Protocol开始说起。下图是Protocol的整个继承结构，从前面我们介绍的一些经验来看，我们先来看一下AbstractProtocol这个抽象接口。
关于AbstractProtocol该接口没有直接实现export和refer方法，该接口主要实现destroy方法以及提供一些公共字段以及公共能力，首先我们看下核心字段，核心字段主要有三个exporterMap、serverMap以及invokers，exporterMap存储服务集合，serverMap存储ProtocolServer实例，invokers存储引用服务的集合。
//存储暴露除去的服务
protected final DelegateExporterMap exporterMap = new DelegateExporterMap();
//ProtocolServer所有实例
protected final Map serverMap = new ConcurrentHashMap<>();
//服务引用的集合
protected final Set>> invokers = new ConcurrentHashSet>>();
这里和介绍一下exporterMap结构，exporterMap是一个Map结构，Key是通过ProtocolUtils.serviceKey方法构建的唯一key, Exporter也就是我们需要暴露除去服务。关于Key构建是可以理解为一个四层Map，第一层按照group分组，group就是URL中配置的内容，通常可以理解为机房、区域等等；剩下的层在GroupServiceKeyCache中，分别按照 serviceName、serviceVersion、port 进行分组，key最终的结构是serviceGroup/serviceName:serviceVersion:port
private String createServiceKey(String serviceName, String serviceVersion, int port) {
StringBuilder buf = new StringBuilder();
if (StringUtils.isNotEmpty(serviceGroup)) {
buf.append(serviceGroup).append('/');
}
buf.append(serviceName);
if (StringUtils.isNotEmpty(serviceVersion) && !"0.0.0".equals(serviceVersion) && !"*".equals(serviceVersion)) {
buf.append(':').append(serviceVersion);
}
buf.append(':').append(port);
return buf.toString();
}
serverMap存储所有的ProtocolServer，也就是服务端，Key是host和port组成的字符串，从URL中获取，ProtocolServer就是对RemotingServer的简单封装，serverMap的填充发生在具体的实现。
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) {
ProtocolServer server = serverMap.get(key);
//双重锁定
if (server == null) {
synchronized (this) {
server = serverMap.get(key);
if (server == null) {
serverMap.put(key, createServer(url));
}
}
} else {
// server supports reset, use together with override
server.reset(url);
}
}
}
invokers主要用于存储被引用的集合，
public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException {
optimizeSerialization(url);
// create rpc invoker.
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
invokers.add(invoker);
return invoker;
}
AbstractProtocol唯一实现的方法就是destory方法，首先会遍历Invokers集合，销毁全部的服务引用，然后遍历全部的exporterMap集合，销毁发布出去的服务。
public void destroy() {
for (Invoker invoker : invokers) {
if (invoker != null) {
//移除所有的引用
invokers.remove(invoker);
try {
if (logger.isInfoEnabled()) {
logger.info("Destroy reference: " + invoker.getUrl());
}
invoker.destroy();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
for (Map.Entry> item : exporterMap.getExporterMap().entrySet()) {
//销毁发布出去的服务
if (exporterMap.removeExportMap(item.getKey(), item.getValue())) {
try {
if (logger.isInfoEnabled()) {
logger.info("Unexport service: " + item.getValue().getInvoker().getUrl());
}
item.getValue().unexport();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
}
DubboProtocol
再开始介绍DubboProtocol之前我们来聊下看源码的另外一个方式，该方式也就是通过单元测试，对于像Dubbo这种优秀的框架，自身的单元测试的覆盖率是比较高的，此外在一些我们疑惑的地方，我们就可以使用单元测试来解决下疑惑，该种方式非常便捷，接下来我们会使用下该方法。 首先我们来看下export方法实现，该方法核心主要就是2个方法:
将invoker转化为DubboExporter，放入exporterMap缓存； 启动ProtocolServer；
public Exporter export(Invoker invoker) throws RpcException {
URL url = invoker.getUrl();
//创建Service key
String key = serviceKey(url);
//将invoker转化为DubboExporter
DubboExporter exporter = new DubboExporter(invoker, key, exporterMap);
//记录到exporterMap
exporterMap.addExportMap(key, exporter);
//export an stub service for dispatching event
Boolean isStubSupportEvent = url.getParameter(STUB_EVENT_KEY, DEFAULT_STUB_EVENT);
Boolean isCallbackservice = url.getParameter(IS_CALLBACK_SERVICE, false);
if (isStubSupportEvent && !isCallbackservice) {
String stubServiceMethods = url.getParameter(STUB_EVENT_METHODS_KEY);
if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
if (logger.isWarnEnabled()) {
logger.warn(new IllegalStateException("consumer [" + url.getParameter(INTERFACE_KEY) +
"], has set stubproxy support event ,but no stub methods founded."));
}
}
}
//启动ProtocolServer
openServer(url);
//序列化优化处理 该方法就是提前将被序列化的类加载到Dubbo中
optimizeSerialization(url);
return exporter;
}
DubboExporter
DubboExporter该类会将invoker进行封装，首先我们来看一下Exporter整体的继承结构，如下图:
在DubboExporter创建时候调用父类AbstractExporter的构造函数，
public DubboExporter(Invoker invoker, String key, DelegateExporterMap delegateExporterMap) {
super(invoker);
this.key = key;
this.delegateExporterMap = delegateExporterMap;
}
在AbstractExporter中存在两个字段invoker和unexported，unexported表示服务是否被销毁，此外该类也对Exporter接口进行实现，在销毁Invoker对象的时候会判断服务的状态，然后在调用destroy进行销毁，afterUnExport方法会执行子类具体的实现，在DubboExporter是移除exporterMap中的缓存的对象。
private final Invoker invoker;
private volatile boolean unexported = false;
@Override
public Invoker getInvoker() {
return invoker;
}
@Override
final public void unexport() {
if (unexported) {
return;
}
unexported = true;
getInvoker().destroy();
afterUnExport();
}
服务端初始化
openServer方法是我们关键方法，该方法会将下层的Exchange、Transport层的方法进行调用，并最终创建NettyServer，此处我们也会使用调试的方式来搞清楚整个调用过程，openServer方法首先判断是否是服务端，然后判断服务是否创建，没有则创建ProtocolServer，否则进行服务重置更新。createServer的时候通过Exchangers门面模式创建，最终封装成为DubboProtocolServer。
private void openServer(URL url) {
// find server.
String key = url.getAddress();
//判断是否为服务端
boolean isServer = url.getParameter(IS_SERVER_KEY, true);
if (isServer) {
ProtocolServer server = serverMap.get(key);
//双重锁定
if (server == null) {
synchronized (this) {
server = serverMap.get(key);
if (server == null) {
serverMap.put(key, createServer(url));
}
}
} else {
// server supports reset, use together with override
server.reset(url);
}
}
}
private ProtocolServer createServer(URL url) {
url = URLBuilder.from(url)
//ReadOnly请求是否阻塞等待
.addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())
//心跳间隔
.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT))
//Codec2扩展实现
.addParameter(CODEC_KEY, DubboCodec.NAME)
.build();
//获取服务端实现 默认是netty
String str = url.getParameter(SERVER_KEY, DEFAULT_REMOTING_SERVER);
//检查服务端扩展实现是否支持
if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
throw new RpcException("Unsupported server type: " + str + ", url: " + url);
}
ExchangeServer server;
try {
//通过Exchangers门面类创建ExchangeServer
server = Exchangers.bind(url, requestHandler);
} catch (RemotingException e) {
throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
}
//检测客户端服务实现是否支持
str = url.getParameter(CLIENT_KEY);
if (str != null && str.length() > 0) {
Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
if (!supportedTypes.contains(str)) {
throw new RpcException("Unsupported client type: " + str);
}
}
//将ExchangeServer包装为DubboProtocolServer
return new DubboProtocolServer(server);
}
在前面我们讲过Transport的设计，对于Exchange是Transport的上层，也就是和Protocol进行交互的，今天我们就从这里来分析Exchange以及Transport调用的整个过程，这样大家就更加理解了Dubbo服务暴露的整个过程，
接下来调用链比较长，我们直接通过单元测试来梳理清楚整个调用链，我们先来查看下export被调用的地方，如下图，我们可以看到该方法被很多地方调用，应为我们是在DubboProtocol类下的方法，因此我们直接使用DubboProtocolTest类下的单元测试就可以。
DubboProtocolTest类下面有很多单测的方法如下图，从名字我们我就可以看出和我们相关应该就是testDemoProtocol和testGetDubboProtocol，这两个方法我们看断言上面来说的话testGetDubboProtocol方法最符合我们的使用，因此我们使用该单元测试。
@Test
public void testGetDubboProtocol(){
DemoService service = new DemoServiceImpl();
int port = NetUtils.getAvailablePort();
protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName())));
Assertions.assertTrue(DubboProtocol.getDubboProtocol().getServers().size() > 0);
}
我们直接将断点放到createServer方法内部，我们可以看到构建URL为，Transporter使用的NettyTransporter，编解码器默认采用DubboCodec。
接下来我们断点放入到Exchangers类的bind方法中，该类采用SPI加载Exchanger，通过调试我们可以发现，最终是采用的是HeaderExchanger，
在HeaderExchanger类中创建HeaderExchangeServer，HeaderExchangeServer该类会创建心跳检测服务，服务端初始化核心的代码在Transporters中，getTransporter方法采用SPI的自适应拓展类，在运行时动态选择NettyTransporter作为实现，
public static RemotingServer 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();
}
接下来我们看一下NettyTransporter类，在该类中直接创建NettyServer；
public class NettyTransporter implements Transporter {
public static final String NAME = "netty";
@Override
public RemotingServer bind(URL url, ChannelHandler handler) throws RemotingException {
return new NettyServer(url, handler);
}
@Override
public Client connect(URL url, ChannelHandler handler) throws RemotingException {
return new NettyClient(url, handler);
}
}
在NettyServer调用父类的AbstractServer，这部分内容我们在通信模块中已经讲过，这里我们就是要将这部分调用的串联起来；
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
// you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
// the handler will be wrapped: MultiMessageHandler->HeartbeatHandler->handler
super(ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME), ChannelHandlers.wrap(handler, url));
}
在AbstractServer中，会调用NettyServer的doOpen方法，用来完成NettyServer的启动；
public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
//调用父类
super(url, handler);
//从URL获取本地地址
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);
//连接数
this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS);
try {
doOpen();
if (logger.isInfoEnabled()) {
logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
}
} catch (Throwable t) {
throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName()
+ " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
}
//创建该服务对应的线程池
executor = executorRepository.createExecutorIfAbsent(url);
}
NettyServer的启动就是Netty的常规的使用，启动过程中要注意下NettyServerHandler，关于该Handler作用就是当服务消费者调用服务提供者的服务时，提供者用来处理各个消息事件，在整一套的调用链上会形成下图的结构，关于这部分内容我们使用一个章节来详细介绍一下，至此就完成整个服务端的启动，最后就会包装成为DubboProtocolServer。
protected void doOpen() throws Throwable {
//创建ServerBootstrap
bootstrap = new ServerBootstrap();
//创建boss EventLoopGroup
bossGroup = NettyEventLoopFactory.eventLoopGroup(1, "NettyServerBoss");
//创建worker EventLoopGroup
workerGroup = NettyEventLoopFactory.eventLoopGroup(
getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
"NettyServerWorker");
//创建一个Netty的ChannelHandler
final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this);
//此处的Channel是Dubbo的Channel
channels = nettyServerHandler.getChannels();
//会话保持
boolean keepalive = getUrl().getParameter(KEEP_ALIVE_KEY, Boolean.FALSE);
bootstrap.group(bossGroup, workerGroup)
.channel(NettyEventLoopFactory.serverSocketChannelClass())
.option(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
.childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
.childOption(ChannelOption.SO_KEEPALIVE, keepalive)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.childHandler(new ChannelInitializer() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// FIXME: should we use getTimeout()?
//连接空闲超时时间
int idleTimeout = UrlUtils.getIdleTimeout(getUrl());
//创建Netty实现的decoder和encoder
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
if (getUrl().getParameter(SSL_ENABLED_KEY, false)) {
//如果配置HTTPS 要实现SslHandler
ch.pipeline().addLast("negotiation",
SslHandlerInitializer.sslServerHandler(getUrl(), nettyServerHandler));
}
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("encoder", adapter.getEncoder())
//心跳检查
.addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS))
//注册nettyServerHandler
.addLast("handler", nettyServerHandler);
}
});
// bind
ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
//等待绑定完成
channelFuture.syncUninterruptibly();
channel = channelFuture.channel();
}
结束
欢迎大家点点关注，点点赞!
