Dubbo-服务暴露

前言

Dubbo源码阅读分享系列文章，欢迎大家关注点赞

SPI实现部分

1. Dubbo-SPI机制
2. Dubbo-Adaptive实现原理
3. Dubbo-Activate实现原理
4. Dubbo SPI-Wrapper

注册中心

5. Dubbo-聊聊注册中心的设计
6. Dubbo-时间轮设计

通信

7. Dubbo-聊聊通信模块设计

RPC

8. 聊聊Dubbo协议

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<String, ProtocolServer> serverMap = new ConcurrentHashMap<>();

//服务引用的集合
protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<Invoker<?>>();

这里和介绍一下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<String, Exporter<?>> 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个方法:

1. 将invoker转化为DubboExporter，放入exporterMap缓存；
2. 启动ProtocolServer；

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
  URL url = invoker.getUrl();

  //创建Service key
  String key = serviceKey(url);
  //将invoker转化为DubboExporter
  DubboExporter<T> exporter = new DubboExporter<T>(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<T> 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<T> invoker;

private volatile boolean unexported = false;

@Override
public Invoker<T> 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<SocketChannel>() {
    @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();

}

image.png

结束

欢迎大家点点关注，点点赞!