首先服务提供方需要定义接口,
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();*/
}
修改服务权重,用来上线服务。
至此,就完成了全部的服务注册和发布的过程。