RPC框架Pigeon简析(二)-- 服务注册与发布

首先服务提供方需要定义接口,

EchoService.java
package com.xx.pigeon.demo;
    public interface EchoService {
        public String echo(String name);
    }

同时,服务提供方同时需要实现该接口,然后服务提供方就可以注册服务,传统的spring注册方法如下:

<bean class="com.dianping.dpsf.spring.ServiceRegistry"
    init-method="init">
      <property name="services">
        <map>
            <entry key="http://service.xx.com/demoService/echoService_1.0.0"
    value-ref="echoServiceImpl" />
        </map>
      </property>
    </bean>

OK,然后我们启动spring上下文的时候,spring会帮助我们初始化这个bean然后调用init方法,这个方法调用了

ServiceFactory.addServices(providerConfigList);

ServiceFactory这个类是个工厂类,它会帮我们初始化服务。剩下就是RPC框架帮我们做的一些事情了。其核心的方法是:

ServicePublisher.addService(providerConfig);
ServerConfig serverConfig = ProviderBootStrap.startup(providerConfig);
ServicePublisher.publishService(providerConfig, false);

其中ServicePublisher.addService(providerConfig) 这个方法主要处理了serviceName,methodName,parameters,同时对服务进行缓存。

ServerConfig serverConfig = ProviderBootStrap.startup(providerConfig)则是启动了“容器”。pigeon目前支持两层,http和tcp。http方面,pigeon会启动内置的jetty,提供了一些服务控制的方法,比如publish、unpublish、online、offline等,同时支持用http来进行RPC。tcp方面,pigeon底层依赖netty进行,所谓启动容器,这里是启动对特定端口的监听。另外,这个方法还会对请求处理器进行注册,这个后面会聊到。

ServicePublisher.publishService(providerConfig, false);该方法为核心方法,主要�用于服务发布。

   if (existingService) {
            boolean autoPublishEnable = ConfigManagerLoader.getConfigManager().getBooleanValue(
                    Constants.KEY_AUTOPUBLISH_ENABLE, true);
            if (autoPublishEnable || forcePublish) {
                List<Server> servers = ProviderBootStrap.getServers(providerConfig);
                int registerCount = 0;
                for (Server server : servers) {
                    publishService(url, server.getRegistryUrl(url), server.getPort(), providerConfig.getServerConfig()
                            .getGroup(), providerConfig.isSupported());//注册服务到zk
                    registerCount++;
                }
                if (registerCount > 0) {
                    boolean isHeartbeatEnable = configManager.getBooleanValue(Constants.KEY_HEARTBEAT_ENABLE,
                            DEFAULT_HEARTBEAT_ENABLE);
                    if (isHeartbeatEnable) {
                        HeartBeatListener.registerHeartBeat(providerConfig);//注册心跳上报
                    }

                    boolean isNotify = configManager
                            .getBooleanValue(Constants.KEY_NOTIFY_ENABLE, DEFAULT_NOTIFY_ENABLE);
                    if (isNotify && serviceChangeListener != null) {
                        serviceChangeListener.notifyServicePublished(providerConfig);//向pigeon的管理中心上报发布情况
                    }

                    boolean autoRegisterEnable = ConfigManagerLoader.getConfigManager().getBooleanValue(
                            Constants.KEY_AUTOREGISTER_ENABLE, true);
                    if (autoRegisterEnable) {
                        ServiceOnlineTask.start();//上线服务
                    } else {
                        logger.info("auto register is disabled");
                    }

                    providerConfig.setPublished(true);
                }
            } else {
                logger.info("auto publish is disabled");
            }
        }

首先会调用方法publishService(url, server.getRegistryUrl(url), server.getPort(), providerConfig.getServerConfig()
.getGroup(), providerConfig.isSupported()),该方法会先计算weight(权重),一般会初始化为0,然后再进行服务注册,

RegistryManager.getInstance().registerService(registryUrl, group, serverAddress, weight);//注册服务地址和服务名
RegistryManager.getInstance().registerSupportNewProtocol(serverAddress, registryUrl, support);//注册服务协议

        if (weight >= 0) {
            if (!serverWeightCache.containsKey(serverAddress)) {
                RegistryManager.getInstance().setServerApp(serverAddress, configManager.getAppName());//注册应用名,一个实例注册一次
                RegistryManager.getInstance().setServerVersion(serverAddress, VersionUtils.VERSION);//注册服务版本,一个实例注册一次
            }
            serverWeightCache.put(serverAddress, weight);
        }

这几个registry其实大同小异,都是将特定的值写入zk,以注册服务为例,

void registerPersistentNode(String serviceName, String group, String serviceAddress, int weight)
            throws RegistryException {
        String weightPath = Utils.getWeightPath(serviceAddress);
        String servicePath = Utils.getServicePath(serviceName, group);
        try {
            if (client.exists(servicePath, false)) {
                Stat stat = new Stat();
                String addressValue = client.get(servicePath, stat);
                String[] addressArray = addressValue.split(",");
                List<String> addressList = new ArrayList<String>();
                for (String addr : addressArray) {
                    addr = addr.trim();
                    if (addr.length() > 0 && !addressList.contains(addr)) {
                        addressList.add(addr.trim());
                    }
                }
                if (!addressList.contains(serviceAddress)) {
                    addressList.add(serviceAddress);
                    Collections.sort(addressList);
                    client.set(servicePath, StringUtils.join(addressList.iterator(), ","), stat.getVersion());
                }
            } else {
                client.create(servicePath, serviceAddress);
            }
            if (weight >= 0) {
                client.set(weightPath, "" + weight);
            }
            if (logger.isInfoEnabled()) {
                logger.info("registered service to persistent node: " + servicePath);
            }
        } catch (Throwable e) {
            if(e instanceof BadVersionException || e instanceof NodeExistsException) {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException ie) {
                    //ignore
                }
                registerPersistentNode(serviceName, group, serviceAddress, weight);
            } else {
                logger.error("failed to register service to " + servicePath, e);
                throw new RegistryException(e);
            }

        }
    }

该方法就是将服务地址和服务名写入zk。其中这里有一个Group的概念,其实就是将同一个服务不同的实例可以分到不同的swimlane里面。
当然RegistryManager可以有不同的实现,你也可以选择将服务信息写到别的地方,比如db。默认实现是写入zk,这也是大型分布式系统比较常用的方法。
然后是注册心跳上报,这个后面会具体谈到。接着,向pigeon的管理中心上报发布情况,这个其实在整个服务中,不是必须的。
最后就是服务上线,从前面的代码我们可以知道服务的权重一般会被初始化为0,而客户端在调用服务的时候会利用权重来负载均衡,也就是说权重是0的服务基本不会被调用到,那么就可以理解为通过修改权重来进行上线或者下线服务。

public void run() {
    logger.info("Service online task start");
    try {
        Thread.sleep(delay);
        if (!isStop) {
            ServiceFactory.online();
        }
    } catch (Throwable e) {
        logger.error(e.getMessage(), e);
    }
}

启动是利用另外一个线程进行的,可以设置一个时延。

public static void online() throws RegistryException {
    logger.info("online");
    ServicePublisher.setServerWeight(Constants.WEIGHT_DEFAULT);
    /*ServiceProviderFactory.notifyServiceOnline();*/
}

修改服务权重,用来上线服务。

至此,就完成了全部的服务注册和发布的过程。

    原文作者:FutureTech
    原文地址: https://www.jianshu.com/p/fdd59cf8a8ce
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞