Dubbo-服务暴露

前言

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

SPI实现部分

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

注册中心

  1. Dubbo-聊聊注册中心的设计
  2. Dubbo-时间轮设计

通信

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

RPC

  1. 聊聊Dubbo协议

AbstractProtocol

在介绍RPC核心接口的时候我们说过Protocol核心作用是将Invoker服务暴露出去以及引用服务将Invoker对象返回,因此我们就从Protocol开始说起。下图是Protocol的整个继承结构,从前面我们介绍的一些经验来看,我们先来看一下AbstractProtocol这个抽象接口。
Dubbo-服务暴露插图
关于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
Dubbo-服务暴露插图1

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  Invoker protocolBindingRefer(Class serviceType, URL url) throws RpcException {
  optimizeSerialization(url);

// create rpc invoker.
DubboInvoker invoker = new DubboInvoker(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个方法:

  1. 将invoker转化为DubboExporter,放入exporterMap缓存;
  2. 启动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整体的继承结构,如下图:
Dubbo-服务暴露插图2
在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 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服务暴露的整个过程,
Dubbo-服务暴露插图3
接下来调用链比较长,我们直接通过单元测试来梳理清楚整个调用链,我们先来查看下export被调用的地方,如下图,我们可以看到该方法被很多地方调用,应为我们是在DubboProtocol类下的方法,因此我们直接使用DubboProtocolTest类下的单元测试就可以。
Dubbo-服务暴露插图4
DubboProtocolTest类下面有很多单测的方法如下图,从名字我们我就可以看出和我们相关应该就是testDemoProtocol和testGetDubboProtocol,这两个方法我们看断言上面来说的话testGetDubboProtocol方法最符合我们的使用,因此我们使用该单元测试。
Dubbo-服务暴露插图5

@Test
public void testGetDubboProtocol(){
  DemoService service = new DemoServiceImpl();
  int port = NetUtils.getAvailablePort();
  protocol.export(proxy.getInvoker(service, DemoService.classURL.valueOf("dubbo://127.0.0.1:" + port + "/" + DemoService.class.getName())));
                                   Assertions.assertTrue(DubboProtocol.getDubboProtocol().getServers().size() > 0);
}

我们直接将断点放到createServer方法内部,我们可以看到构建URL为,Transporter使用的NettyTransporter,编解码器默认采用DubboCodec。
Dubbo-服务暴露插图6
接下来我们断点放入到Exchangers类的bind方法中,该类采用SPI加载Exchanger,通过调试我们可以发现,最终是采用的是HeaderExchanger,
Dubbo-服务暴露插图7
在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(00, idleTimeout, MILLISECONDS))
        //注册nettyServerHandler
        .addLast("handler", nettyServerHandler);
    }
  });
// bind
ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
//等待绑定完成
channelFuture.syncUninterruptibly();
channel = channelFuture.channel();

}
Dubbo-服务暴露插图8
image.png

结束

欢迎大家点点关注,点点赞!
Dubbo-服务暴露插图9

文章来源于互联网:Dubbo-服务暴露

THE END
分享
二维码