接口定义
@SPI("netty") //缺省值netty
public interface Transporter {
/**
* Bind a server.
*
* @param url erver url
* @param handler
* @return server
* @throws RemotingException
* @see com.alibaba.dubbo.remoting.Transporters#bind(URL, ChannelHandler...)
*url参数 含有server 或者transporter
*/
@Adaptive({Constants.SERVER_KEY, Constants.TRANSPORTER_KEY})
Server bind(URL url, ChannelHandler handler) throws RemotingException;
/**
* Connect to a server.
*
* @param url server url
* @param handler
* @return client
* @throws RemotingException
* @see com.alibaba.dubbo.remoting.Transporters#connect(URL, ChannelHandler...)
* url参数含有server 或者transporter
*/
@Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})
Client connect(URL url, ChannelHandler handler) throws RemotingException;
}
类图
说明
接上一篇《dubbo源码阅读-远程暴露(七)之Exchangers》
@Override public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { //1. handler 会再次经过2层包装，增加功能 ChannelHandler h = new DecodeHandler(new HeaderExchangeHandler(handler)); //<1>2. Transports 操作会启动netty 监听端口，配置序列化实现， Server s = Transporters.bind(url,h); /** * <6>HeaderExchangeServer： 会对nettyServer 进行包装， 主要增加2个功能： * * a. 对channel进行 空闲时间检测，超过则关闭连接，节省资源。 * * b. 如果server关闭，则发送消息给client端，不再发送请求到该server。 */ return new HeaderExchangeServer( s); }
Transporters
<1>bind
com.alibaba.dubbo.remoting.Transporters#bind(com.alibaba.dubbo.common.URL, com.alibaba.dubbo.remoting.ChannelHandler...)
/** * @param url * @param handlers 经过DecodeHandler->HeaderExchangeHandler->com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol#requestHandler逐级装饰 * @return * @throws RemotingException */ 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 { //如果传入多个handles 通过ChannelHandlerDispatcher包装 内部通过包装使每个Handle都能监听各个事件 handler = new ChannelHandlerDispatcher(handlers); } /** * <2>getTransporter * <3>bind */ return getTransporter().bind(url, handler); }
<2>getTransporter
com.alibaba.dubbo.remoting.Transporters#getTransporter
public static Transporter getTransporter() { //SPI扩展点 缺省值是 @SPI("netty") return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension(); }
NettyTransporter
/** * 经过DecodeHandler->HeaderExchangeHandler->com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol#requestHandler逐级装饰 * @param url erver url * @param listener * @return * @throws RemotingException */ @Override public Server bind(URL url, ChannelHandler listener) throws RemotingException { //<3>创建一个nettyServer 此处handle是 return new NettyServer(url, listener); }
<3>NettyServer
com.alibaba.dubbo.remoting.transport.netty4.NettyServer#NettyServer
/** * <4>ChannelHandlers.wrap * <6>super * @param url * @param handler * @throws RemotingException */ public NettyServer(URL url, ChannelHandler handler) throws RemotingException { super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME))); }
<4>ChannelHandlers.wrap
public static ChannelHandler wrap(ChannelHandler handler, URL url) { //<5>打包 return ChannelHandlers.getInstance().wrapInternal(handler, url); }
<5>wrapInternal
protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) { //将handler经过MultiMessageHandler->HeartbeatHandler->SPI获取Dispatcher默认是ALL 进行装饰 return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class) .getAdaptiveExtension().dispatch(handler, url))); }
<7>doOpen
com.alibaba.dubbo.remoting.transport.netty4.NettyServer#doOpen
@Override protected void doOpen() throws Throwable { /** * boss线程，主要监听端口和获取worker线程及分配socketChannel给worker * NamedThreadFactory自定义线程工厂 并设置线程为守护线程 主线程关闭 子线程自动关闭 */ bootstrap = new ServerBootstrap(); // worker线程负责数据读写 bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true)); //创建niosocket工厂 workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS), new DefaultThreadFactory("NettyServerWorker", true)); /** * NettyServerHandler->NettyServer->...-ProtocolRequestHandle * f * */ final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this); channels = nettyServerHandler.getChannels(); //参数讲解:https://blog.csdn.net/zhongzunfa/article/details/94590670 bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)//tcp协议 禁用Nagle 算法 .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE)//tcp协议 允许通一个端口可以绑定到多个套接字 .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)//tcp协议 .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);//处理器 } }); // 启动服务 ChannelFuture channelFuture = bootstrap.bind(getBindAddress()); channelFuture.syncUninterruptibly(); channel = channelFuture.channel(); }
<6>AbstractServer
public AbstractServer(URL url, ChannelHandler handler) throws RemotingException { super(url, handler); /** * 从url获取绑定ip和端口如:dubbo://127.0.0.1:23888.... * localAddress例子:/192.168.2.1:23888 * 成员变量 */ localAddress = getUrl().toInetSocketAddress(); //获取绑定id优先从url参数bind.ip获取 获取不到取url里面封装的 String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost()); //获取绑定端口 优先从url bind.port 获取获取不到取url封装的 int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort()); //anyhost是否有取到绑定ip 参见:https://www.cnblogs.com/LQBlog/p/12469007.html#autoid-6-12-0 if (url.getParameter(Constants.ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) { bindIp = NetUtils.ANYHOST; } //初始化绑定address 成员变量 bindAddress = new InetSocketAddress(bindIp, bindPort); /** * 获取url accepts 控制客户端连接服务端的连接数控制 默认0 * 文档地址:http://dubbo.apache.org/zh-cn/docs/user/references/xml/dubbo-protocol.html * 成员变量 */ this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS); /** * 获取url参数 idle.timeout 空闲超时时间，单位：毫秒 * 成员变量 */ this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT); try { //<7>开启服务器 模板方法模式 由子类实现 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); } //获得线程池 DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension(); executor = (ExecutorService) dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY, Integer.toString(url.getPort())); }