spring cloud eureka client
2018-12-22 本文已影响15人
靈08_1024
本文是在搭建好eureka服务的基础上，来进行分析的。还有姊妹篇eureka server。
eureka Client工作的过程
大体工作过程，分为以下几个阶段，如下图所示：
eureka Client工作
DiscoveryClient初步了解
服务在读取配置信息，有一个特殊的客户端接口org.springframework.cloud.client.discovery.DiscoveryClient，可以通过注入这个接口来获取一些客户端配置。
@RestController
public class ServiceInstanceRestController {
@Autowired
private DiscoveryClient discoveryClient;
/**
* 填入任一EurekaClient的实例名，便可获取该服务的相关信息
*
* @param applicationName EurekaClient的appName对应spring.application.name属性
* @return
*/
@RequestMapping("service-instances/{applicationName}")
public List<ServiceInstance> serviceInstancesByApplication(@PathVariable String applicationName) {
return discoveryClient.getInstances(applicationName);
}
/**
* 获取当前服务的描述
*
* @return
*/
@RequestMapping("description")
public String description() {
return discoveryClient.description();
}
/**
* 获取所有的服务名
*
* @return
*/
@RequestMapping("getServices")
public List<String> getServices() {
return discoveryClient.getServices();
}
}
源码和配置
我们都知道Eureka是netfix的杰作，在了解了上述的spring的DiscoveryClient后，我们来看看com.netflix.discovery.DiscoveryClient.java这个类，他也是Eureka Client的核心聚焦类。
client主要类图.png
该类封装了注册、心跳等一系列行为。所以，下面的代码分析，我们从注册和心跳开始。
在EurekaClientConfig.java中，该类封装了Client与Server交互的配置信息。其中有两个比较重要的属性：
fetchRegistry为true表示该Client从Server中拉取注册信息，对应的配置为eureka.client.fetch-registry；
registerWithEureka为true表示该Client是否注册到Server上，对应的配置为eureka.client. register-with-eureka。
这两个值会用在DiscoveryClient中，若都为false，表示既不服务发现，也不服务注册。
在下面会定义一个调度线程池，大小为2，一个是用于心跳，一个用于缓存刷新。
scheduler = Executors.newScheduledThreadPool(2,
new ThreadFactoryBuilder()
.setNameFormat("DiscoveryClient-%d")
.setDaemon(true)
.build());
heartbeatExecutor = new ThreadPoolExecutor(
1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
new ThreadFactoryBuilder()
.setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
.setDaemon(true)
.build()
); // use direct handoff
cacheRefreshExecutor = new ThreadPoolExecutor(
1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
new ThreadFactoryBuilder()
.setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
.setDaemon(true)
.build()
); // use direct handoff
全量拉取注册表
下面为DicoveryClient#getAndStoreFullRegistry方法
private void getAndStoreFullRegistry() throws Throwable {
//获取当前版本
long currentUpdateGeneration = fetchRegistryGeneration.get();
Applications apps = null;
//发送请求
EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == null
? eurekaTransport.queryClient.getApplications(remoteRegionsRef.get())
: eurekaTransport.queryClient.getVip(clientConfig.getRegistryRefreshSingleVipAddress(), remoteRegionsRef.get());
if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
//取请求体
apps = httpResponse.getEntity();
}
if (apps == null) {
logger.error("The application is null for some reason. Not storing this information");
} else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
//版本比较，重新过滤和洗牌apps，并将UP状态的Application放入本地区域apps中
localRegionApps.set(this.filterAndShuffle(apps));
logger.debug("Got full registry with apps hashcode {}", apps.getAppsHashCode());
} else {
logger.warn("Not updating applications as another thread is updating it already");
}
}
其中，从eureka拉取所有Application信息的请求是{{eureka-server}}/eureka/apps。
因为该方法可以同时被多个线程访问，是线程不安全的（会产生各种覆盖现象）。所以其中的关键值 localRegionApps为本地区域应用，采用AtomicReference<Applications>类型；fetchRegistryGeneration为当前client存储的注册表版本，类型为AtomicLong。这两个值都是使用原子类型来确保其安全。具体使用，参见上面源码。
该方法的输出结果（样例，便于后面方法的了解）：
<applications>
<versions__delta>1</versions__delta>
<apps__hashcode>UP_5_</apps__hashcode>
<application>
<name>ADMIN-CLIENT</name>
<instance>
<instanceId>admin-client2</instanceId>
<hostName>localhost</hostName>
<app>ADMIN-CLIENT</app>
<ipAddr>192.168.1.105</ipAddr>
<status>UP</status>
<overriddenstatus>UNKNOWN</overriddenstatus>
<port enabled="true">8082</port>
<securePort enabled="false">443</securePort>
<countryId>1</countryId>
<dataCenterInfo class="com.netflix.appinfo.InstanceInfo$DefaultDataCenterInfo">
<name>MyOwn</name>
</dataCenterInfo>
<leaseInfo>
<renewalIntervalInSecs>30</renewalIntervalInSecs>
<durationInSecs>90</durationInSecs>
<registrationTimestamp>1546006066759</registrationTimestamp>
<lastRenewalTimestamp>1546006066759</lastRenewalTimestamp>
<evictionTimestamp>0</evictionTimestamp>
<serviceUpTimestamp>1546006066759</serviceUpTimestamp>
</leaseInfo>
<metadata>
<management.port>8082</management.port>
<jmx.port>63368</jmx.port>
</metadata>
<homePageUrl>http://localhost:8082/</homePageUrl>
<statusPageUrl>http://localhost:8082/actuator/info</statusPageUrl>
<healthCheckUrl>http://localhost:8082/actuator/health</healthCheckUrl>
<vipAddress>admin-client</vipAddress>
<secureVipAddress>admin-client</secureVipAddress>
<isCoordinatingDiscoveryServer>false</isCoordinatingDiscoveryServer>
<lastUpdatedTimestamp>1546006066759</lastUpdatedTimestamp>
<lastDirtyTimestamp>1546006040724</lastDirtyTimestamp>
<actionType>ADDED</actionType>
</instance>
<instance>
<instanceId>admin-client1</instanceId>
...
</instance>
</application>
<application>
<name>EUREKA-SERVICE</name>
<instance>
...
</instance>
<instance>
...
</instance>
</application>
<application>
<name>ADMIN-SERVER</name>
<instance>
...
</instance>
</application>
</applications>
Tips：Eureka Instance的运行状态有5种：
- UP：存活，
- DOWN ：不再发送心跳的，
- STARTING ： 启动中，
- OUT_OF_SERVICE ：关闭了的，
- UNKNOWN ：未知服务。
增量拉取注册表
下面为DicoveryClient#getAndUpdateDelta方法
private void getAndUpdateDelta(Applications applications) throws Throwable {
long currentUpdateGeneration = fetchRegistryGeneration.get();
Applications delta = null;
EurekaHttpResponse<Applications> httpResponse = eurekaTransport.queryClient.getDelta(remoteRegionsRef.get());
if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
delta = httpResponse.getEntity();
}
if (delta == null) {
......
} else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
String reconcileHashCode = "";
//使用重入锁机制来更新增量注册信息
if (fetchRegistryUpdateLock.tryLock()) {
try {
//更新本地缓存
updateDelta(delta);
//计算一致性hashCode
reconcileHashCode = getReconcileHashCode(applications);
} finally {
fetchRegistryUpdateLock.unlock();
}
} else {
logger.warn("Cannot acquire update lock, aborting getAndUpdateDelta");
}
if (!reconcileHashCode.equals(delta.getAppsHashCode()) || clientConfig.shouldLogDeltaDiff()) {
//比较版本，CAS更新。发起远程调用线程
reconcileAndLogDifference(delta, reconcileHashCode);
}
} else {
logger.warn("Not updating application delta as another thread is updating it already");
logger.debug("Ignoring delta update with apps hashcode {}, as another thread is updating it already", delta.getAppsHashCode());
}
}
其中，从eureka server拉取增量Application信息的请求是{{eureka-server}}/eureka/apps/delta。
本段代码使用重入锁机制来确保多线程操作，防止数据污染。
其一致性hashCode举例为：UP_3_形式的。
服务注册
在拉取完注册表后，client会注册自己到server中去。DiscoveryClient#register方法。
boolean register() throws Throwable {
EurekaHttpResponse<Void> httpResponse;
try {
httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
} catch (Exception e) {
logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e);
throw e;
}
return httpResponse.getStatusCode() == 204;
}
其中，注册client到server的请求是{{eureka-server}}/eureka/apps/{{client-instant-appName}}。
响应码为204表示成功。
服务续约（心跳）
boolean renew() {
EurekaHttpResponse<InstanceInfo> httpResponse;
try {
httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
if (httpResponse.getStatusCode() == Status.NOT_FOUND.getStatusCode()) {
REREGISTER_COUNTER.increment();
long timestamp = instanceInfo.setIsDirtyWithTime();
boolean success = register();
if (success) {
instanceInfo.unsetIsDirty(timestamp);
}
return success;
}
return httpResponse.getStatusCode() == Status.OK.getStatusCode();
} catch (Throwable e) {
logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e);
return false;
}
}
其中，从client续约到server的请求是{{eureka-server}}/eureka/apps/{{client-instance-appName}}/{client-instance-id}。
