订阅服务看起来是比较简单容易实现的功能，乍一看，就是发送订阅服务的名称给注册中心，然后注册中心反馈给服务消费者，感觉万事大吉，其实并不是这样的，这块是比较容易实现的，使用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，这边的代码逻辑还是相对比较复杂的，详细看源码
说也不是说的很清楚，还是看代码来的实在~还请见谅