1、暴露服务到远程
上一篇文章分析了暴露服务到本地，6、Dubbo的服务导出1之导出到本地。接下来分析暴露服务到远程。
if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) { if (registryURLs != null && !registryURLs.isEmpty()) { for (URL registryURL : registryURLs) { // 添加动态参数,此动态参数是决定Zookeeper创建临时节点还是持久节点 url = url.addParameterIfAbsent(Constants.DYNAMIC_KEY, registryURL.getParameter(Constants.DYNAMIC_KEY)); String proxy = url.getParameter(Constants.PROXY_KEY); if (StringUtils.isNotEmpty(proxy)) { registryURL = registryURL.addParameter(Constants.PROXY_KEY, proxy); } // 步骤1)创建Invoker,这里创建Invoker逻辑和上面一样 Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded( Constants.EXPORT_KEY, url.toFullString())); DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this); // 步骤2)暴露服务 Exporter<?> exporter = ServiceConfig.protocol.export(wrapperInvoker); exporters.add(exporter); } }} /** * 下面分析步骤2,该方法两大核心逻辑,导出服务和注册服务,服务注册下篇文章分析 */@Overridepublic <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { // 1. 导出服务,export invoker,本篇文章仅分析第一步导出服务到远程 final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker); // zookeeper://10.101.99.127:2181/com.alibaba.dubbo.registry.RegistryService // ?application=demo-provider&dubbo=2.0.2 URL registryUrl = getRegistryUrl(originInvoker); // registry provider,默认返回ZookeeperRegistry实例 final Registry registry = getRegistry(originInvoker); // dubbo://172.22.213.93:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true // &application=demo-provider&default.server=netty4&dubbo=2.0.2&generic=false // &interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=8140&side=provider final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker); // 不配置的话默认返回true boolean register = registeredProviderUrl.getParameter("register", true); ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl); // 2.注册服务,这篇文章已经比较长了,决定将步骤2和步骤3新起一篇文章分析,服务暴露之后需要注册到注册中心 if (register) { register(registryUrl, registeredProviderUrl); ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true); } // 3.数据更新订阅 final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registeredProviderUrl); final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker); overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener); registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener); return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registeredProviderUrl);} 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)); // 调用protocol的export方法导出服务,默认是采用Dubbo协议,对应DubboProtocol的export方法 // 但是这里protocol.export()并不是先走DubboProtocol的export方法,而是先走 // ProtocolListenerWrapper的wrapper方法 // 因为ProtocolListenerWrapper对DubboProtocol做了一层包装,具体参考 // https://segmentfault.com/a/1190000020387196,核心方法protocal.export() exporter = new ExporterChangeableWrapper<T>( (Exporter<T>) protocol.export(invokerDelegete), originInvoker); bounds.put(key, exporter); } } } return exporter;}/** * 上述核心方法protocol.export()会先走到ProtocolListenerWrapper的export方法,该方法是在服务暴露上做了 监听器功能的增强,也就是加上了监听器 */@Overridepublic <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { // 如果是注册中心,则暴露该invoker if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } // 创建一个暴露者监听器包装类对象,暴露服务时这里的protocol是ProtocolFilterWrapper,这里用到了 // Wrapper包装原有的DubboProtocol,可以参考https://segmentfault.com/a/1190000020387196 return new ListenerExporterWrapper<T>(protocol.export(invoker), Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class) .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY)));} /** * ProtocolFilterWrapper的export方法,该方法是在服务暴露上做了过滤器链的增强,也就是加上了过滤器 */@Overridepublic <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { // 如果是注册中心,则直接暴露服务 if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } // 服务提供侧暴露服务,这里通过buildInvokerChain形成了过滤器链 return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));}/** * 该方法就是创建带Filter链的Invoker对象,倒序的把每一个过滤器串连起来,形成一个invoker */private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) { Invoker<T> last = invoker; // 获得过滤器的所有扩展实现类实例集合 List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class). getActivateExtension(invoker.getUrl(), key, group); if (!filters.isEmpty()) { // 从最后一个过滤器开始循环,创建一个带有过滤器链的invoker对象 for (int i = filters.size() - 1; i >= 0; i--) { final Filter filter = filters.get(i); final Invoker<T> next = last; last = new Invoker<T>() { @Override public Class<T> getInterface() { return invoker.getInterface(); } @Override public URL getUrl() { return invoker.getUrl(); } @Override public boolean isAvailable() { return invoker.isAvailable(); } // 关键在这里,调用下一个filter代表的invoker,把每一个过滤器串起来 @Override public Result invoke(Invocation invocation) throws RpcException { return filter.invoke(next, invocation); } @Override public void destroy() { invoker.destroy(); } @Override public String toString() { return invoker.toString(); } }; } } return last;}// 经过两个Wrapper的export方法包装之后,走到DubboProtocol的export方法,这里是核心方法public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { // url形如dubbo://172.22.213.93:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true // &application=demo-provider&bind.ip=172.22.213.93&bind.port=20880&dubbo=2.0.2&generic=false // /&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=648&qos.port=22222 // &side=provider×tamp=1569585915258 URL url = invoker.getUrl(); // 获取服务标识,理解成服务坐标也行,由服务组名,服务名,服务版本号以及端口组成,key形如 // com.alibaba.dubbo.demo.DemoService:20880 String key = serviceKey(url); // 创建DubboExporter DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap); // 将<key, exporter>键值对放入缓存中 exporterMap.put(key, exporter); // 本地存根相关代码, export an stub service for dispatching event // 删除,暂时还没有分析本地存根相关 // 启动服务器,重点关注这里 openServer(url); optimizeSerialization(url); return exporter;}// 根据URL值可以猜测,openServer方法就是启动Netty服务器,在172.22.213.93:20880端口上监听调用请求openServer(url); /** * 在同一台机器上(单网卡),同一个端口上仅允许启动一个服务器实例,若某个端口上已有服务器实例,此时则调用reset方法 重置服务器的一些配置 */private void openServer(URL url) { // 获取host:port,并将其作为服务器实例的key,用于标识当前的服务器实例,key形如172.22.213.93:20880 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); if (server == null) { // 创建服务器实例,put之后serverMap形如<172.22.213.93:20880, HeaderExchangeServer> serverMap.put(key, createServer(url)); } else { // 服务器已创建,则根据url中的配置重置服务器 server.reset(url); } }} private ExchangeServer createServer(URL url) { url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString()); // 添加心跳检测配置到URL中,enable heartbeat by default url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT)); // 获取server参数,默认为netty,这里配置成了netty4,str就为netty4 String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER); // 通过SPI检测是否存在server参数所代表的Transporter拓展,不存在则抛出异常 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, DubboCodec.NAME); ExchangeServer server; try { // 创建 ExchangeServer,核心方法 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;}public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange"); return getExchanger(url).bind(url, handler);}public static Exchanger getExchanger(URL url) { // 默认type就是header String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER); // 创建HeadExchanger return getExchanger(type);}public static Exchanger getExchanger(String type) { return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);} public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { // 用传输层绑定返回的server创建对应的信息交换服务端 // 这里也是分成两步,下面先分析bind方法,该方法就是开启Netty4服务器监听请求 // 1) bind方法 // 2) new HeaderExchangeServer(Server server) return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));}// 步骤1)bind方法public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException { ChannelHandler handler; if (handlers.length == 1) { handler = handlers[0]; } else { handler = new ChannelHandlerDispatcher(handlers); } // 获取自适应Transporter实例,并调用实例方法. getTransporter()方法获取的Transporter是在运行时动态创建的, // 类名为TransporterAdaptive,也就是自适应拓展类.TransporterAdaptive会在运行时根据传入的URL参数决定加载 // 什么类型的Transporter,这里我们配置成了netty4的NettyTransporter // String string = url.getParameter("server", url.getParameter("transporter", "netty")); // transporter = ExtensionLoader.getExtensionLoader(Transporter.class).getExtension(string); return getTransporter().bind(url, handler); }public Server bind(URL url, ChannelHandler listener) throws RemotingException { // 创建一个NettyServer return new NettyServer(url, listener);}public NettyServer(URL url, ChannelHandler handler) throws RemotingException { // 调用父类构造方法,这里的wrap方法返回的是 // MultiMessageHandler->HeartbeatHandler->AllDispatcherHandler->DecodeHandler->HeaderExchangeHandler // -> 表示前一个handler里面包装了下一个handler super(url, ChannelHandlers.wrap(handler,ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));}public static ChannelHandler wrap(ChannelHandler handler, URL url) { return ChannelHandlers.getInstance().wrapInternal(handler, url);}// 包装了MultiMessageHandler功能,增加了多消息处理的功能,以及对心跳消息做了功能增强protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) { // 调用了多消息处理器,对心跳消息进行了功能加强 return new MultiMessageHandler( new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class) .getAdaptiveExtension().dispatch(handler, url)));}public AbstractServer(URL url, ChannelHandler handler) throws RemotingException { // url形如dubbo://172.22.213.93:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true // &application=demo-provider&bind.ip=172.22.213.93&bind.port=20880&channel.readonly.sent=true // &codec=dubbo&default.server=netty4&dubbo=2.0.2&generic=false&heartbeat=60000 // &interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=6900&qos.port=22222 // &side=provider×tamp=1569633535398 // handler是MultiMessageHandler实例 super(url, handler); // 从url中获得本地地址,/172.22.213.93:20880 localAddress = getUrl().toInetSocketAddress(); // 从url配置中获得绑定的ip,本机IP地址172.22.213.93 String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost()); // 从url配置中获得绑定的端口号,20880 int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort()); // 判断url中配置anyhost是否为true或者判断host是否为不可用的本地Host,url中配置了anyhost为true if (url.getParameter(Constants.ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) { bindIp = NetUtils.ANYHOST; } // /0.0.0.0:20880 bindAddress = new InetSocketAddress(bindIp, bindPort); // 从url中获取配置,默认值为0 this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS); // 从url中获取配置,默认600s this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT); try { // 开启服务器 doOpen(); } DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension(); executor = (ExecutorService) dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY, Integer.toString(url.getPort()));} /** * 该类是端点的抽象类,其中封装了编解码器以及两个超时时间. * 基于dubbo 的SPI机制,获得相应的编解码器实现对象,编解码器优先从Codec2的扩展类中寻找 */public AbstractEndpoint(URL url, ChannelHandler handler) { super(url, handler); this.codec = getChannelCodec(url); // 优先从url配置中取,如果没有,默认为1s this.timeout = url.getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT); // 优先从url配置中取,如果没有,默认为3s this.connectTimeout = url.getPositiveParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT);}// 该方法是创建服务器,并且开启@Overrideprotected void doOpen() throws Throwable { // 创建服务引导类 bootstrap = new ServerBootstrap(); // 创建boss线程组 bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true)); // 创建worker线程组 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(); // 设置eventLoopGroup还有可选项 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)) .addLast("decoder", adapter.getDecoder()) .addLast("encoder", adapter.getEncoder()) // 加入了NettyServerHandler,后续消息处理应该就是通过这个来处理,猜测,TODO .addLast("handler", nettyServerHandler); } }); // bind绑定,这里bind完成之后Netty服务器就启动了,监听20880端口上的请求,有兴趣可以研究下Netty的源码 ChannelFuture channelFuture = bootstrap.bind(getBindAddress()); // 等待绑定完成 channelFuture.syncUninterruptibly(); // 设置通道 channel = channelFuture.channel();}// 步骤2)new HeaderExchangeServer(Server server)// 构造函数就是对属性的设置,心跳的机制以及默认值都跟HeaderExchangeClient中的一模一样public HeaderExchangeServer(Server server) { if (server == null) { throw new IllegalArgumentException("server == null"); } this.server = server; //获得心跳周期配置,如果没有配置,默认设置为0 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();} /** * 该方法是开始心跳,跟HeaderExchangeClient类中的开始心跳方法唯一区别是获得的通道不一样, * 客户端跟通道是一一对应的,所有只要对一个通道进行心跳检测,而服务端跟通道是一对多的关系, * 所有需要对该服务器连接的所有通道进行心跳检测 */private void startHeartbeatTimer() { // 先停止现有的心跳检测 stopHeartbeatTimer(); if (heartbeat > 0) { // 创建心跳定时器 heartbeatTimer = scheduled.scheduleWithFixedDelay( new HeartBeatTask(new HeartBeatTask.ChannelProvider() { @Override public Collection<Channel> getChannels() { // 返回一个不可修改的连接该服务器的信息交换通道集合 return Collections.unmodifiableCollection( HeaderExchangeServer.this.getChannels()); } }, heartbeat, heartbeatTimeout), heartbeat, heartbeat, TimeUnit.MILLISECONDS); }}/** * 该类实现了Runnable接口,实现的是心跳任务,里面包含了核心的心跳策略 */final class HeartBeatTask implements Runnable { // 通道管理 private ChannelProvider channelProvider; // 心跳间隔,单位:ms private int heartbeat; // 心跳超时时间,单位:ms private int heartbeatTimeout; HeartBeatTask(ChannelProvider provider, int heartbeat, int heartbeatTimeout) { this.channelProvider = provider; this.heartbeat = heartbeat; this.heartbeatTimeout = heartbeatTimeout; } /** * 该方法中是心跳机制的核心逻辑,注意以下几个点： * * 如果需要心跳的通道本身如果关闭了,那么跳过,不添加心跳机制. * 无论是接收消息还是发送消息,只要超过了设置的心跳间隔,就发送心跳消息来测试是否断开 * 如果最后一次接收到消息到到现在已经超过了心跳超时时间,那就认定对方的确断开,分两种情况来处理对方断开的情况. * 分别是服务端断开,客户端重连以及客户端断开,服务端断开这个客户端的连接.这里要好好品味一下谁是发送方, * 谁在等谁的响应,苦苦没有等到. */ @Override public void run() { try { long now = System.currentTimeMillis(); // 遍历所有通道 for (Channel channel : channelProvider.getChannels()) { // 如果通道关闭了,则跳过 if (channel.isClosed()) { continue; } try { // 最后一次接收到消息的时间戳 Long lastRead = (Long) channel.getAttribute( HeaderExchangeHandler.KEY_READ_TIMESTAMP); // 最后一次发送消息的时间戳 Long lastWrite = (Long) channel.getAttribute( HeaderExchangeHandler.KEY_WRITE_TIMESTAMP); // 如果最后一次接收或者发送消息到时间到现在的时间间隔超过了心跳间隔时间 if ((lastRead != null && now - lastRead > heartbeat) || (lastWrite != null && now - lastWrite > heartbeat)) { // 创建一个request,设置版本号,设置需要得到响应 Request req = new Request(); req.setVersion(Version.getProtocolVersion()); req.setTwoWay(true); // 设置事件类型,为心跳事件 req.setEvent(Request.HEARTBEAT_EVENT); // 发送心跳请求 channel.send(req); } // 如果最后一次接收消息的时间到现在已经超过了超时时间 if (lastRead != null && now - lastRead > heartbeatTimeout) { // 如果该通道是客户端,也就是请求的服务器挂掉了,客户端尝试重连服务器 if (channel instanceof Client) { try { // 重新连接服务器 ((Client) channel).reconnect(); } catch (Exception e) { //do nothing } } else { // 如果不是客户端,也就是是服务端返回响应给客户端,但是客户端挂掉了, // 则服务端关闭客户端连接 channel.close(); } } } } } }interface ChannelProvider { // 获得所有的通道集合,需要心跳的通道数组 Collection<Channel> getChannels();}}