Netty在Dubbo服务暴露时何时被使用

书唐瑞

发布于 2022-06-02 14:06:01

7250

发布于 2022-06-02 14:06:01

文章被收录于专栏：Netty历险记

Dubbo的底层通信使用的是Netty.

关于Dubbo的服务暴露流程,网络上已经有很多优质的文章.此篇文章以Dubbo的服务暴露为主线(不会详细讲解),观察一下,Netty在服务暴露过程中何时被使用.

// 服务暴露的起点
com.alibaba.dubbo.config.spring.ServiceBean#onApplicationEvent

@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
    if (isDelay() && !isExported() && !isUnexported()) {
        if (logger.isInfoEnabled()) {
            logger.info("The service ready on spring started. service: " + getInterface());
        }
        // 服务暴露
        export();
    }
}

流程会走到如下代码

private void doExportUrls() {
  List<URL> registryURLs = loadRegistries(true);
  for (ProtocolConfig protocolConfig : protocols) {
    doExportUrlsFor1Protocol(protocolConfig, registryURLs);
  }
}

registryURLs的值如下(样例)
registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=infuq-dubbo-provider&check=false&dubbo=2.0.2&pid=4916&registry=zookeeper&timestamp=1609052431702

protocols的值如下(样例)
<dubbo:protocol name="dubbo" threads="200" port="20880" id="dubbo" />

流程继续走到如下代码

Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

Exporter<?> exporter = protocol.export(wrapperInvoker);

基于SPI机制,protocol会调用com.alibaba.dubbo.registry.integration.RegistryProtocol#export,源码如下

@Override
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {

  final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);

  URL registryUrl = getRegistryUrl(originInvoker);

  final Registry registry = getRegistry(originInvoker);
  final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);

  boolean register = registeredProviderUrl.getParameter("register", true);

  ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);

  if (register) {
    register(registryUrl, registeredProviderUrl);
    ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
  }
  
  ...
}

第一行的doLocalExport方法,继续跟进

private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) {
  String key = getCacheKey(originInvoker);
  ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
  if (exporter == null) {
    synchronized (bounds) {
      exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
      if (exporter == null) {
        final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
        // 暴露
        exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
        bounds.put(key, exporter);
      }
    }
  }
  return exporter;
}

基于SPI机制,protocol会调用com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol#export,源码如下

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

  ...

  // 开启服务器
  openServer(url);
  optimizeSerialization(url);
  return exporter;
}

继续跟进openServer方法

private void openServer(URL url) {

  String key = url.getAddress();
  boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
  if (isServer) {
    ExchangeServer server = serverMap.get(key);
    if (server == null) {
      // 创建服务器
      serverMap.put(key, createServer(url));
    } else {
      server.reset(url);
    }
  }
}

url如下(样例)
dubbo://192.168.0.102:20880/com.infuq.facade.QueryUserInfoFacade?anyhost=true&application=infuq-dubbo-provider&bean.name=com.infuq.facade.QueryUserInfoFacade&bind.ip=192.168.0.102&bind.port=20880&dubbo=2.0.2&generic=false&interface=com.infuq.facade.QueryUserInfoFacade&methods=update&pid=13528&revision=1.0.0&side=provider&threads=200&timestamp=1609053786543&version=1.0.0

一直跟进,会依次调用如下几个主要方法

server = Exchangers.bind(url, requestHandler);

getExchanger(url).bind(url, handler);

new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));

getTransporter().bind(url, handler);

new NettyServer(url, listener);

在创建NettyServer时候,最终会调用到如下代码

@Override
protected void doOpen() throws Throwable {
  bootstrap = new ServerBootstrap();

  bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true));
  workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS), new DefaultThreadFactory("NettyServerWorker", true));

  final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this);
  channels = nettyServerHandler.getChannels();

  bootstrap.group(bossGroup, workerGroup)
      .channel(NioServerSocketChannel.class)
      .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
      .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
      .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
      .childHandler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel ch) throws Exception {
          NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
          ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
              .addLast("decoder", adapter.getDecoder())
              .addLast("encoder", adapter.getEncoder())
              .addLast("handler", nettyServerHandler);
        }
      });
  // bind
  ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
  channelFuture.syncUninterruptibly();
  channel = channelFuture.channel();

}

上面的代码就是很熟悉的Netty创建服务端的代码.最终会创建并启动服务端.

也就是说,在暴露服务的过程中,在进行doLocalExport本地暴露的时候,会分别经过RegistryProtocol#export和DubboProtocol#export,最后通过Netty创建一个服务端,监听外部的接口调用请求.

虽然本地服务已经暴露,但是还需要将服务注册到注册中心(例如ZK)

在没有注册到ZK之前,查看下ZK信息

是没有dubbo节点信息的.

@Override
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
  // 本地暴露
  final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);

  URL registryUrl = getRegistryUrl(originInvoker);
  final Registry registry = getRegistry(originInvoker);
  final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);

  boolean register = registeredProviderUrl.getParameter("register", true);
  ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);

  if (register) {
    // 注册服务
    register(registryUrl, registeredProviderUrl);
    ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
  }

  ...
}

registryUrl内容如下(样例)
zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=infuq-dubbo-provider&check=false&dubbo=2.0.2&export=dubbo%3A%2F%2F192.168.0.102%3A20880%2Fcom.infuq.facade.QueryUserInfoFacade%3Fanyhost%3Dtrue%26application%3Dinfuq-dubbo-provider%26bean.name%3Dcom.infuq.facade.QueryUserInfoFacade%26bind.ip%3D192.168.0.102%26bind.port%3D20880%26dubbo%3D2.0.2%26generic%3Dfalse%26interface%3Dcom.infuq.facade.QueryUserInfoFacade%26methods%3Dupdate%26pid%3D9892%26revision%3D1.0.0%26side%3Dprovider%26threads%3D200%26timestamp%3D1609055193386%26version%3D1.0.0&pid=9892&timestamp=1609055193386

registeredProviderUrl内容如下(样例)
dubbo://192.168.0.102:20880/com.infuq.facade.QueryUserInfoFacade?anyhost=true&application=infuq-dubbo-provider&bean.name=com.infuq.facade.QueryUserInfoFacade&dubbo=2.0.2&generic=false&interface=com.infuq.facade.QueryUserInfoFacade&methods=update&pid=9892&revision=1.0.0&side=provider&threads=200&timestamp=1609055193386&version=1.0.0

当执行完register方法之后,再查看ZK信息