源码分析Dubbo网络通信篇NettyServer、HeaderExchangeServer
本文主要分析一下NettyServer，HeaderExchangeServer实现细节。 1、NettyServer NettyServer整个类图如下： 首先从全貌上大概看一下NettyServer对象所持有的属性： AbstractPeer private final ChannelHandler handler 事件处理Handler。 private volatile URL url 该协议的第一个服务提供者的URL， Server只需要用到 URL中的参数，与具体某一个服务没什么关系。 AbstractEndpoint private Codec2 codec 编码解码器。 private int timeout 超时时间 private int connectTimeout 连接超时时间 AbstractServer private InetSocketAddress localAddress ：url host:port地址。 private InetSocketAddress bindAddress：如果是多网卡，并且指定了 bind.ip、bind.port，如果为空，与localAddress相同。 private int accepts ： AbstractServer#accepts未使用到。 private int idleTimeout = 600; AbstractServer#accepts未使用到。 NettyServer private Map< String, Channel> channels：< ip:port, channel> 所有通道。 private ServerBootstrap bootstrap ： netty 服务端启动器。 private io.netty.channel.Channel channel：服务端监听通道。 private EventLoopGroup bossGroup;Netty boss线程组（负责连接事件） private EventLoopGroup workerGroup ： nety work线程组（负责IO事件） 1.1 NettyServer 构造方法 public NettyServer(URL url, ChannelHandler handler) throws RemotingException { super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME))); } 直接调用父类的public AbstractServer(URL url, ChannelHandler handler)方法，从前面的文章中得知， ChannelHandlers.wrap方法会对ChannelHandler handler进行封装，主要是加入事件分发模式(Dispatch)。 1.1.1 AbstractServer构造方法 public AbstractServer(URL url, ChannelHandler handler) throws RemotingException { super(url, handler); // @1 localAddress = getUrl().toInetSocketAddress(); // @2 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); // @3 this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS); this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT); // @4 try { doOpen(); // @5 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())); } 代码@1：调用父类的构造方法，主要初始化AbstractPeer（channelHandler、url）和AbstractEndpoint（codec2、timeout、idleTimeout ） 代码@2：根据URL中的host与端口，创建localAddress。 代码@3：如果配置了< dubbo:parameter key = "bind.ip" value = ""/> 与 < dubbo:parameter key = "bind.port" />，则用该IP与端口创建bindAddress，通常用于多网卡，如果未配置，bindAddress与 localAddress绑定的IP与端口一样。 代码@4：初始化accepts与idleTimeout ,这两个参数未被其他地方使用。 代码@5，调用doOpen方法，正式在相应端口建立网络监听。 1.2、源码分析NettyServer#doOpen protected void doOpen() throws Throwable { NettyHelper.setNettyLoggerFactory(); bootstrap = new ServerBootstrap(); // @1 bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true)); // @2 workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS), new DefaultThreadFactory("NettyServerWorker", true)); // @3 final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this); // @4 channels = nettyServerHandler.getChannels(); bootstrap.group(bossGroup, workerGroup) // @5 .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()); // @6 channelFuture.syncUninterruptibly(); channel = channelFuture.channel(); } 代码@1：创建Netty服务端启动帮助类ServerBootstrap. 代码@2：创建服务端Boss线程，线程名：.NettyServerBoss，主要负责客户端的连接事件，主从多Reactor线程模型中的主线程（连接事件）。 代码@3：创建服务端Work线程组，线程名：NettyServerWorker-序号，线程个数取自参数：iothreads，默认为(CPU核数+1)与32取小值，顾名思义，IO线程数，主要处理读写事件，编码、解码都在IO线程中完成。 代码@4：创建用户Handler,这里是NettyServerHandler。 代码@5：Netty启动的常规写法，关注如下内容： addLast("decoder", adapter.getDecoder()) : 添加解码器 addLast("encoder", adapter.getEncoder()) ：添加编码器 addLast("handler", nettyServerHandler) ：添加业务Handler。 这里简单介绍一下流程： 客户端建立与服务端连接，此时Boss线程的连接事件触发，建立TCP连接，并向IO线程注册该通道(Channel0)的读事件。 当客户端向服务端发送请求消息后，IO线程中的读事件触发，会首先调用adapter.getDecoder() 根据对应的请求协议（例如dubbo）从二进制流中解码出一个完整的请求对象，然后传入到业务handler,例如nettyServerHandler，执行相应的事件方法，例如recive方法。 当服务端向Channel写入响应结果时，首先编码器会按照协议编码成二进制流，供客户端解码。 如果对Netty想深入学习的话，请移步到作者的《源码分析Netty系列》 2、HeaderExchangeServer 根据 Dubbo 服务端初始化流程，我们可知，Dubbo 为了封装各种不同的网络实现客户端（netty、mina）等，引入了 Exchangers 层，存在 ExchangeServer，其实现 Server 并内部持有具体的 Server 实现端，例如 NettyServer。 接下来，我们重点来关注一下 HeaderExchangeServer. 核心属性如下： ScheduledExecutorService scheduled：心跳线程数，线程名称前缀，dubbo-remoting-server-heartbeat-thread-序号 private final Server server：具体的Server实现类，例如NettyServer。 private ScheduledFuture< ?> heartbeatTimer：心跳调度Future，可以通过future取消心跳等动作。 private int heartbeat：心跳间隔时间 private int heartbeatTimeout：心跳超时时间，至少为heartbeat的两倍 2.1 构造函数 public HeaderExchangeServer(Server server) { if (server == null) { throw new IllegalArgumentException("server == null"); } this.server = server; this.heartbeat = server.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0); this.heartbeatTimeout = server.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3); if (heartbeatTimeout < heartbeat * 2) { throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2"); } startHeartbeatTimer(); } 说明，主要是通过heartbeat参数设置心跳间隔，如果不配置，则不启动心跳检测。从上面看来HeaderExchangeServer内部持有Server,并封装了心跳的功能，在这里就不细细分析了。 >作者介绍：丁威，《RocketMQ技术内幕》作者，RocketMQ 社区优秀布道师、CSDN2019博客之星TOP10,维护公众号：中间件兴趣圈目前已陆续发表源码分析Java集合、Java 并发包(JUC)、Netty、Mycat、Dubbo、RocketMQ、Mybatis等源码专栏。可以点击链接加入中间件知识星球 ，一起探讨高并发、分布式服务架构，交流源码。 </niosocketchannel>
