不灭的焱

革命尚未成功,同志仍须努力

作者:php-note.com  发布于:2018-11-04 20:03  分类:Java库/系统 

订阅服务看起来是比较简单容易实现的功能,乍一看,就是发送订阅服务的名称给注册中心,然后注册中心反馈给服务消费者,感觉万事大吉,其实并不是这样的,这块是比较容易实现的,使用Netty很容易就能实现了,但是写的时候就会发现各种问题

 

比如一个现在系统第一次调用callRemotingService()的方法的时候,这时候在没有任何条件的支持下,服务消费者首先需要去注册中心去获取信息,然后根据拿到的信息去与服务提供者建立Netty的长连接,获取到长连接的channel之后就可以进行调用了

 

感觉就是很简单,跟以前我们处理订阅,或者发布服务的处理都跟类似,不过这边却多了一个建立长连接的过程,这边处理就相对比较复杂一点,因为我们知道Netty建立连接是异步的,这个就像双刃剑一样,你拿着IP和端口号,使用Netty去建立连接的时候,因为异步,代码不会阻塞到连接成功建立后返回,立即返回,然后你就会拿着还没有建立好连接的channel去调用,这样就会导致调用错误

 

虽然这种场景不多见,一般情况下,项目都是使用spring管理的,spring一般会预加载的,如果提前建立好所有的服务的长连接,就可以很好的规避这个问题,不过这个问题却是客观存在的,所以我们需要解决这个问题

 

首先,我们还看看一个基本的服务消费者定义的基本接口吧(代码大部分来自于Jupiter):

 

package org.laopopo.client.consumer;

import io.netty.channel.Channel;

import org.laopopo.common.exception.remoting.RemotingSendRequestException;
import org.laopopo.common.exception.remoting.RemotingTimeoutException;
import org.laopopo.common.loadbalance.LoadBalanceStrategy;
import org.laopopo.common.utils.ChannelGroup;
import org.laopopo.common.utils.UnresolvedAddress;
import org.laopopo.remoting.model.RemotingTransporter;



/**
 * 
 * @author BazingaLyn
 * @description 消费端的接口
 * @time 2016年8月15日
 * @modifytime
 */
public interface Consumer {
	
	/**
	 * 远程调用方法
	 * @param serviceName 远程调用的服务名
	 * @param args 参数
	 * @return 
	 * @throws Throwable
	 */
	Object call(String serviceName,Object... args) throws Throwable;
	
	/**
	 * 
	 * @param serviceName
	 * @param timeout 调用超时时间
	 * @param args
	 * @return
	 * @throws Throwable
	 */
	Object call(String serviceName,long timeout,Object... args) throws Throwable;
	
	/**
	 * 在当服务端向注册中心订阅服务的时候进行管理
	 * @param serviceName
	 * @return
	 */
	SubscribeManager subscribeService(String serviceName);
	
	
	/**
	 * 去连接注册中心,获取到与注册中心的唯一一个channel
	 */
	void getOrUpdateHealthyChannel();
	
	/**
	 * 去注册中心订阅服务,注册中心返回结果之后,回调NotifyListener的方法
	 * @param subcribeServices
	 * @param listener
	 */
	void subcribeService(String subcribeServices,NotifyListener listener);
	
	/**
	 * 
	 * @param address 该地址是提供者的地址
	 * 一个服务提供者的地址可以维护一组channel,因为一个消费者实例与一个提供者之间的长链接数可以不止一个,不过当然一般情况下,一个就可以了
	 * @return
	 */
	ChannelGroup group(UnresolvedAddress address);
	
	/**
	 * 根据一个服务名,匹配用户给这个服务设定的负载均衡策略,根据这个负载均衡算法去找到这个服务对应的与提供者的Channel
	 * @param serviceName
	 * @return
	 */
	ChannelGroup loadBalance(String serviceName);
	
	/**
	 * 当注册中心告之某个服务多了一个提供者之后,我们需要将其更新
	 * @param serviceName
	 * @param group
	 * @return
	 */
	boolean addChannelGroup(String serviceName, ChannelGroup group);
	
	/**
	 * 当注册中心告之某个服务的提供者下线的时候,我们也需要服务路由表
	 * @param serviceName
	 * @param group
	 * @return
	 */
	boolean removeChannelGroup(String serviceName, ChannelGroup group);
	
	/**
	 * 核心方法,远程调用
	 * @param channel 消费者与服务提供者的之间建立的长连接的channel
	 * @param request 请求体 包含请求的参数,请求的方法名
	 * @param timeout 请求超时时间
	 * @return
	 * @throws RemotingTimeoutException
	 * @throws RemotingSendRequestException
	 * @throws InterruptedException
	 */
	RemotingTransporter sendRpcRequestToProvider(Channel channel, RemotingTransporter request,long timeout) throws RemotingTimeoutException, RemotingSendRequestException, InterruptedException;
	
	/**
	 * 当注册中心推送某个服务的负载均衡策略发送变化之后,需要变更的信息
	 * @param serviceName
	 * @param loadBalanceStrategy
	 */
	void setServiceLoadBalanceStrategy(String serviceName,LoadBalanceStrategy loadBalanceStrategy);
	
	
	/**
	 * 如果直连的情况下,根据address直接获取连接
	 * @param address
	 * @return
	 * @throws InterruptedException 
	 */
	Channel directGetProviderByChannel(UnresolvedAddress address) throws InterruptedException;
	
	/**
	 * 启动consumer端实例
	 */
	void start();
	
	
	interface SubscribeManager {

		/**
		 * 启动对订阅服务管理的服务
		 */
        void start();

        /**
         * 当某个服务去注册中心注册之后,注册中心返回订阅结果,consumer实例
         * 拿着订阅结果,去向服务提供者建立长连接,因为建立长连接的过程是异步的,简而言之获取一个active的channel的是异步的,所以当一切貌似看起来ok的时候
         * 其实未必,所以必须进行回调管理,否则远程调用的时候,可能channel还没有准备就绪,会报错
         * 
         * 详细见 {@link NotifyListener} 类的头部注释
         * @param timeoutMillis
         * @return
         */
        boolean waitForAvailable(long timeoutMillis);
    }
	

}

 

因为本小节主要说明订阅这个操作,所以我们着重看SubscribeManager subscribeService(String serviceName)这个方法:

 

@Override
public SubscribeManager subscribeService(final String service) {

	SubscribeManager manager = new SubscribeManager() {

		private final ReentrantLock lock = new ReentrantLock();
		private final Condition notifyCondition = lock.newCondition();
		private final AtomicBoolean signalNeeded = new AtomicBoolean(false);

		@Override
		public void start() {
			subcribeService(service, new NotifyListener() {
				
				@Override
				public void notify(RegisterMeta registerMeta, NotifyEvent event) {

					// host
					String remoteHost = registerMeta.getAddress().getHost();
					// port vip服务 port端口号-2
					int remotePort = registerMeta.isVIPService() ? (registerMeta.getAddress().getPort() - 2) : registerMeta.getAddress().getPort();

					final ChannelGroup group = group(new UnresolvedAddress(remoteHost, remotePort));
					if (event == NotifyEvent.CHILD_ADDED) {
						// 链路复用,如果此host和port对应的链接的channelGroup是已经存在的,则无需建立新的链接,只需要将此group与service建立关系即可
						if (!group.isAvailable()) {

							int connCount = registerMeta.getConnCount() < 0 ? 1 : registerMeta.getConnCount();

							group.setWeight(registerMeta.getWeight());

							for (int i = 0; i < connCount; i++) {

								try {
									// 所有的consumer与provider之间的链接不进行短线重连操作
									DefaultConsumer.this.getProviderNettyRemotingClient().setreconnect(false);
									DefaultConsumer.this.getProviderNettyRemotingClient().getBootstrap()
											.connect(ConnectionUtils.string2SocketAddress(remoteHost + ":" + remotePort)).addListener(new ChannelFutureListener() {

												@Override
												public void operationComplete(ChannelFuture future) throws Exception {
													group.add(future.channel());
													onSucceed(signalNeeded.getAndSet(false));
												}
												
											});
								} catch (Exception e) {
									logger.error("connection provider host [{}] and port [{}] occor exception [{}]", remoteHost, remotePort, e.getMessage());
								}
							}
						}else{
							onSucceed(signalNeeded.getAndSet(false));
						}
						addChannelGroup(service,group);
					}else if(event == NotifyEvent.CHILD_REMOVED){
						removedIfAbsent(service, group);
					}
				}
			});
		}

		@Override
		public boolean waitForAvailable(long timeoutMillis) {
			if (isServiceAvailable(service)) {
                return true;
            }
			boolean available = false;
            long start = System.nanoTime();
            final ReentrantLock _look = lock;
            _look.lock();
            try {
                while (!isServiceAvailable(service)) {
                    signalNeeded.set(true);
                    notifyCondition.await(timeoutMillis, MILLISECONDS);

                    available = isServiceAvailable(service);
                    if (available || (System.nanoTime() - start) > MILLISECONDS.toNanos(timeoutMillis)) {
                        break;
                    }
                }
            } catch (InterruptedException e) {
                JUnsafe.throwException(e);
            } finally {
                _look.unlock();
            }
            return available;
		}

		private boolean isServiceAvailable(String service) {
			CopyOnWriteArrayList<ChannelGroup> list = DefaultConsumer.super.getChannelGroupByServiceName(service);
			if(list == null){
				return false;
			}else{
				for(ChannelGroup channelGroup : list){
					if(channelGroup.isAvailable()){
						return true;
					}
				}
			}
			return false;
		}

		private void onSucceed(boolean doSignal) {
			if (doSignal) {
                final ReentrantLock _look = lock;
                _look.lock();
                try {
                    notifyCondition.signalAll();
                } finally {
                    _look.unlock();
                }
            }
		}

	};
	manager.start();
	return manager;
}

 

这段代码主要的核心意思就是获取的信息之后,建立连接,因为连接是异步的,所以加了通知机制,建立成功之后,将健康的channel放入到group中去,然后就维护了所有的channel,这边的代码逻辑还是相对比较复杂的,详细看源码

 

说也不是说的很清楚,还是看代码来的实在~还请见谅

 

 

摘自:https://blog.csdn.net/linuu/article/details/52776081