Luckylau's Blog

微服务架构之Dubbo服务发布(3)

发表于 | 分类于 |

netty的启动

接着微服务架构之Dubbo服务发布(2）我们详细讲解

1
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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));
//这里最重要的是调用DubboProtocol的export方法，导出完之后，返回一个新的ExporterChangeableWrapper实例
                    exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
                    bounds.put(key, exporter);
                }
            }
        }
        return exporter;
    }

DubboProtocol的外面包裹着ProtocolFilterWrapper，再外面还包裹着ProtocolListenerWrapper。会先经过ProtocolListenerWrapper。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
//DubboProtocol
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        URL url = invoker.getUrl();
        
        // export service.
  //key由serviceName，version，portgroup,组成
    //当nio客户端发起远程调用时，nio服务端通过此key来决定调用哪个Exporter，也就是执行的Invoker。
        String key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
  //缓存
        exporterMap.put(key, exporter);
  //是否要支持本地存根 ，可以用stub关键字配置，这也是dubbo服务降级方式，详情见服务降级（2）
        //export an stub service for dispaching event
        Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY,Constants.DEFAULT_STUB_EVENT);
        Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
        if (isStubSupportEvent && !isCallbackservice){
            String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
            if (stubServiceMethods == null || stubServiceMethods.length() == 0 ){
                if (logger.isWarnEnabled()){
                    logger.warn(new IllegalStateException("consumer [" +url.getParameter(Constants.INTERFACE_KEY) +
                            "], has set stubproxy support event ,but no stub methods founded."));
                }
            } else {
                stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
            }
        }
//根据URL绑定IP与端口，建立NIO框架的Server
        openServer(url);
        // modified by lishen
        optimizeSerialization(url);
        return exporter;
}

exporter会被放到bounds中，key就是上面生成的，客户端发请求就根据key找到对应的Exporter，转化为invoker进行调用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
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(Constants.IS_SERVER_KEY, true);
    if (isServer) {
        ExchangeServer server = serverMap.get(key);
      //同一JVM中，同协议的服务，共享同一个Server，
      //第一个暴露服务的时候创建server，
      //以后相同协议的服务都使用同一个server
        if (server == null) {
            serverMap.put(key, createServer(url));
        } else {
      // server supports reset, use together with override
      //HeaderExchangeServer处理的，主要是重置如果url带有accepts，threads，heartbeat，heartbeat.timeout等参数信息。
            server.reset(url);
        }
    }
}

我们来看看createServer（URL url）具体实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
private ExchangeServer createServer(URL url) {
        //默认开启server关闭时发送readonly事件
        url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
        //默认开启heartbeat
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
  //默认使用netty
        String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);
        if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
            throw new RpcException("Unsupported server type: " + str + ", url: " + url);
        url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
        ExchangeServer server;
        try {
          //Exchanger默认只有一个实现HeaderExchanger,负责数据交换和网络通信
            server = Exchangers.bind(url, requestHandler);
        } catch (RemotingException e) {
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }
        str = url.getParameter(Constants.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);
            }
        }
        return server;
    }

Exchangers.bind(url, requestHandler)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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);
    }
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
    }//HeaderExchangeServer实例化时候会启动心跳计数器

接着看Transporters.bind(URL url, ChannelHandler… handlers)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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);
        }
     //getTransporter()获取一个Adaptive的Transporter
    //然后调用bind方法（默认是NettyTransporter的bind方法）
        return getTransporter().bind(url, handler);
    }

NettyTransporter的bind方法,目前支持netty和netty4,最后的super都指向了AbstractServer构造方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
//netty4
public Server bind(URL url, ChannelHandler listener) throws RemotingException {
    return new NettyServer(url, listener);
}
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
    super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}
//netty
public Server bind(URL url, ChannelHandler listener) throws RemotingException {
    return new NettyServer(url, listener);
}
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
    super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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(Constants.ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
            bindIp = NetUtils.ANYHOST;
        }
        bindAddress = new InetSocketAddress(bindIp, bindPort);
        this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);
        this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);
        try {
            doOpen();//最后的关键，也区分netty和netty4
            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);
        }
        //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()));
    }

最后看看doOpen()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
//netty
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(Constants.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.setPipelineFactory(new ChannelPipelineFactory() {
         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());
 }
 //netty4
 protected void doOpen() throws Throwable {
     NettyHelper.setNettyLoggerFactory();
     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();
 }

参考

http://blog.csdn.net/jurson99/article/details/50401270

https://www.jianshu.com/p/dcfe426e9cd7

Luckylau wechat
如果对您有价值,看官可以打赏的！