Pigeon服务的注册与发现

上一篇：Pigeon的一次调用服务端发生了什么
前面介绍了Pigeon请求调用的过程，那么Pigeon的服务是如何被调用方感知的呢，我们知道通常RPC都会有一个注册中心，以保存服务测的信息（ip，port等等），这样客户端就可以通过注册中心去拿到服务的信息，来完成请求的调用。那一起来看下Pigeon服务的注册和发现机制是如何实现的。

服务的注册

首先，我们来看下服务的注册机制。
在Pigeon中主要使用了@Reference和@Service注解，用来标志服务调用方接口和服务提供方实现类，而对这两个注解进行拦截处理的类是AnnotationBean类，这里是通过实现了Spring的BeanPostProcessor接口来完成处理的（Pigeon相关的@Service类扫描和其BeanDefinition的封装实通过实现了BeanFactoryPostProcessor接口实现的），看下postProcessAfterInitialization方法实现：

public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {// 先判断要注入相关属性的Bean是否在指定的注解扫描包下Class<?> beanClass = AopUtils.getTargetClass(bean);if (beanClass == null || !isMatchPackage(beanClass.getName())) {return bean;}// 判断类定义中是否存在@Service注解Service service = beanClass.getAnnotation(Service.class);if (service != null) {// 如果未自定义接口，则用当前beanClassClass serviceInterface = service.interfaceClass();if (void.class.equals(service.interfaceClass())) {serviceInterface = ServiceConfigUtils.getServiceInterface(beanClass);}if (serviceInterface == null) {serviceInterface = beanClass;}// 初始化ProviderConfig和ServerConfig，完成服务提供者配置和服务器配置的初始化ProviderConfig<Object> providerConfig = new ProviderConfig<Object>(serviceInterface, bean);providerConfig.setService(bean);providerConfig.setUrl(service.url());providerConfig.setVersion(service.version());providerConfig.setSharedPool(service.useSharedPool());providerConfig.setActives(service.actives());ServerConfig serverConfig = new ServerConfig();serverConfig.setPort(getDefaultPort(service.port()));serverConfig.setSuffix(service.group());serverConfig.setAutoSelectPort(service.autoSelectPort());providerConfig.setServerConfig(serverConfig);// 注册服务提供者，启动服务器，发布服务，完成pigeon提供方调用初始化ServiceFactory.addService(providerConfig);}// 解析bean中方法和属性是否包含Reference，完成bean作为服务调用方的依赖注入。postProcessBeforeInitialization(bean, beanName);return bean;}

这里我们关系的是注册过程，即ServiceFactory.addService()，进去看下它的最终触发的实现

public void doAddService(ProviderConfig providerConfig) {try {// 检查服务名checkServiceName(providerConfig);// 发布指定版本服务，同时解析服务方法ServicePublisher.addService(providerConfig);// 启动netty RPC服务器，作为服务提供方供调用方调用服务ServerConfig serverConfig = ProviderBootStrap.startup(providerConfig);// 更新serverConfigproviderConfig.setServerConfig(serverConfig);// 实际发布服务，会将服务注册到注册中心，供调用方发现调用ServicePublisher.publishService(providerConfig, false);} catch (Throwable t) {throw new RpcException("error while adding service:" + providerConfig, t);}}

其中在添加解析服务中，主要进行了以下流程：
判断是否配置了版本，如果配置了，生成带版本的urlWithVersion，更新key=urlWithVersion的服务，同时如果大于key=url的对应服务版本，会用新版本覆盖默认url版本
如果服务实现了InitializingService接口，调用实现的initialize方法
调用ServiceMethodFactory.init(url)方法，用来初始化调用ServiceMethodFactory的ServiceMethodCache：遍历ServicePublisher的所有服务提供类，建立ServiceMethodCache，存储该类下所有满足要求的方法和方法id的映射关系
首先会忽略掉Object和Class的所有方法
过滤方法后，判断是否需要压缩，根据url+"#"+方法名的方式进行hash。
代码如下

public static <T> void addService(ProviderConfig<T> providerConfig) throws Exception {if (logger.isInfoEnabled()) {logger.info("add service:" + providerConfig);}String version = providerConfig.getVersion();String url = providerConfig.getUrl();// 默认版本，直接以url为keyif (StringUtils.isBlank(version)) {serviceCache.put(url, providerConfig);} else {// urlWithVersion = url + "_" + versionString urlWithVersion = getServiceUrlWithVersion(url, version);if (serviceCache.containsKey(url)) {// 如果已经存在，覆盖服务serviceCache.put(urlWithVersion, providerConfig);ProviderConfig<?> providerConfigDefault = serviceCache.get(url);String defaultVersion = providerConfigDefault.getVersion();// 如果默认服务存在默认版本，并且小于当前版本，用当前版本服务更新默认服务版本if (!StringUtils.isBlank(defaultVersion)) {if (VersionUtils.compareVersion(defaultVersion, providerConfig.getVersion()) < 0) {serviceCache.put(url, providerConfig);}}} else {// 将当前版本设为指定版本服务和默认版本服务serviceCache.put(urlWithVersion, providerConfig);// use this service as the default providerserviceCache.put(url, providerConfig);}}// 如果服务实现了InitializingService接口，调用实现的initialize方法T service = providerConfig.getService();if (service instanceof InitializingService) {((InitializingService) service).initialize();}// 解析接口自定义方法，根据方法名，参数等相关信息记录方法ServiceMethodFactory.init(url);
}// ServiceMethodFactory.init(url);方法实现如下：
public static void init(String url) {getServiceMethodCache(url);
}
// 具体调用了
public static ServiceMethodCache getServiceMethodCache(String url) {// 是否存在指定url的ServiceMethodCacheServiceMethodCache serviceMethodCache = methods.get(url);if (serviceMethodCache == null) {// 获取指定url的providerConfigMap<String, ProviderConfig<?>> services = ServicePublisher.getAllServiceProviders();ProviderConfig<?> providerConfig = services.get(url);if (providerConfig != null) {Object service = providerConfig.getService();Method[] methodArray = service.getClass().getMethods();serviceMethodCache = new ServiceMethodCache(url, service);// 遍历指定url的所有服务方法for (Method method : methodArray) {// 忽略掉Object和Class的所有方法if (!ingoreMethods.contains(method.getName())) {method.setAccessible(true);serviceMethodCache.addMethod(method.getName(), new ServiceMethod(service, method));if (isCompact) {// 压缩url，方法名等调用所需信息int id = LangUtils.hash(url + "#" + method.getName(), 0, Integer.MAX_VALUE);ServiceId serviceId = new ServiceId(url, method.getName());ServiceId lastId = CompactRequest.PROVIDER_ID_MAP.putIfAbsent(id, serviceId);// 检查如果存在相同id服务方法，抛异常if (lastId != null && !serviceId.equals(lastId)) {throw new IllegalArgumentException("same id for service:" + url + ", method:"+ method.getName());}}}}// 更新缓存methods.put(url, serviceMethodCache);}}return serviceMethodCache;
}

服务的启动主要是指netty服务端接口监听开启，请求处理线程池初始化及启动（处理请求调用链的线程池），代码如下：

这里我们重点看下服务的发布方法ServicePublisher.publishService主要逻辑

如果服务需要注册，这里调用了publishServiceToRegistry方法去注册服务，通过在ZK上创建一个持久节点完成注册，然后通知服务变化。

服务的发现

我们来看下调用方又是如何获得服务节点信息的

服务节点信息获取是在调用方获取serviceProxy动态代理时完成的，客户端动态代理的实现我们前面一篇已经说过，这里只看发现的部分

这里实现了对客户端的注册和服务发现

这一步主要将调用方相关信息注册到注册中心中，大致实现如下
1.获取服务地址列表，这里主要从注册中心zk获取，具体获取地址如/DP/SERVER/com.dianping.pigeon.demo.EchoService，获取到的实际值是ip:port，如172.23.51.30:6354，如果有多个ip:port，会以逗号分割
遍历服务列表，从注册红心获取每个服务地址的权重，如通过/DP/WEIGHT/172.23.51.30:6354获取，获取到ip,port,weight后，会封装成一个HostInfo
2.通过RegistryEventListener发布providerAdded和serverInfoChanged事件，对于providerAdded事件：
会导致ClientManager.InnerServiceProviderChangeListener触发进行客户端注册，首先会初始化一个ConnectInfo，包含serviceName, host, port, weight等信息，然后会进行以下两步：
进一步会触发相关的ClusterListener调用addConnect，这里主要包含：
触发NettyClientFactory#createClient调用，进一步创建NioClientSocketChannelFactory，建立和服务端的netty连接，并建立心跳检测
触发WeightFactorMaintainer#providerAdded，进一步调用addWeight方法，更新weights和weightFactors成员。
在RegistryManager中添加服务引用地址，更新referencedServiceAddresses和referencedAddresses
触发WeightFactorMaintainer#providerAdded，进一步调用addWeight方法，更新weights和weightFactors成员。这里和前面一步有重复

public Set<HostInfo> registerClients(InvokerConfig invokerConfig) {String remoteAppkey = invokerConfig.getRemoteAppKey();String serviceName = invokerConfig.getUrl();String group = RegistryManager.getInstance().getGroup(serviceName);String vip = invokerConfig.getVip();logger.info("start to register clients for service '" + serviceName + "#" + group + "'");String localHost = null;if (vip != null && vip.startsWith("console:")) {localHost = configManager.getLocalIp() + vip.substring(vip.indexOf(":"));}// 从注册中心获取服务地址String serviceAddress = getServiceAddress(invokerConfig);String[] addressArray = serviceAddress.split(",");Set<HostInfo> addresses = Collections.newSetFromMap(new ConcurrentHashMap<HostInfo, Boolean>());for (int i = 0; i < addressArray.length; i++) {if (StringUtils.isNotBlank(addressArray[i])) {// addressList.add(addressArray[i]);String address = addressArray[i];int idx = address.lastIndexOf(":");if (idx != -1) {String host = null;int port = -1;try {host = address.substring(0, idx);port = Integer.parseInt(address.substring(idx + 1));} catch (RuntimeException e) {logger.warn("invalid address:" + address + " for service:" + serviceName);}if (host != null && port > 0) {if (localHost != null && !localHost.equals(host + ":" + port)) {continue;}try {// 从注册中心获取相应服务权重int weight = RegistryManager.getInstance().getServiceWeight(address, serviceName, false);addresses.add(new HostInfo(host, port, weight));} catch (Throwable e) {logger.error("error while registering service invoker:" + serviceName + ", address:"+ address + ", env:" + configManager.getEnv(), e);throw new ServiceUnavailableException("error while registering service invoker:"+ serviceName + ", address:" + address + ", env:" + configManager.getEnv(), e);}}} else {logger.warn("invalid address:" + address + " for service:" + serviceName);}}}final String url = serviceName;long start = System.nanoTime();if (enableRegisterConcurrently) {final CountDownLatch latch = new CountDownLatch(addresses.size());// 并发发布providerAdded和serverInfoChanged时间给RegistryEventListenerfor (final HostInfo hostInfo : addresses) {Runnable r = new Runnable() {@Overridepublic void run() {try {RegistryEventListener.providerAdded(url, hostInfo.getHost(), hostInfo.getPort(),hostInfo.getWeight());RegistryEventListener.serverInfoChanged(url, hostInfo.getConnect());} catch (Throwable t) {logger.error("failed to add provider client:" + hostInfo, t);} finally {latch.countDown();}}};registerThreadPool.submit(r);}try {latch.await();} catch (InterruptedException e) {logger.info("", e);}} else {//顺序执行for (final HostInfo hostInfo : addresses) {RegistryEventListener.providerAdded(url, hostInfo.getHost(), hostInfo.getPort(), hostInfo.getWeight());RegistryEventListener.serverInfoChanged(url, hostInfo.getConnect());}}long end = System.nanoTime();logger.info("end to register clients for service '" + serviceName + "#" + group + "', cost:"+ ((end - start) / 1000000));return addresses;}