spring cloud中 eureka部分源码分析及微服务管理功能

内容简介：spring cloud中 eureka部分源码分析及微服务管理功能

一. EurekaClient获取所有注册的服务

eureka客户端会加载一个定时任务去获取注册中心的服务，任务的配置在：com.netflix.discovery.DiscoveryClient，刷新的线程是：CacheRefreshThread。

获取的注册中心服务的时候，会把所有服务都拉取下来，但是默认会过滤掉状态不是UP的服务。

获取服务的具体代码在：DiscoveryClient.getAndStoreFullRegistry()方法

/**

• Gets the full registry information from the eureka server and stores it locally.
• When applying the full registry, the following flow is observed:

*

• if (update generation have not advanced (due to another thread))
• atomically set the registry to the new registry
• fi

*

• @return the full registry information.
• @throws Throwable
• on error.

*/

private void getAndStoreFullRegistry() throws Throwable {

long currentUpdateGeneration = fetchRegistryGeneration.get();

logger.info("Getting all instance registry info from the eureka server");

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();
}
logger.info("The response status is {}", httpResponse.getStatusCode());

if (apps == null) {
    logger.error("The application is null for some reason. Not storing this information");
} else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
    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");
}

}

过滤服务状态的代码在：DiscoveryClient.filterAndShuffle()方法

/**

• Gets the applications after filtering the applications for
• instances with only UP states and shuffling them.

*

• <p>
• The filtering depends on the option specified by the configuration
• {@link EurekaClientConfig#shouldFilterOnlyUpInstances()}. Shuffling helps
• in randomizing the applications list there by avoiding the same instances
• receiving traffic during start ups.
• </p>

*

• @param apps
• The applications that needs to be filtered and shuffled.
• @return The applications after the filter and the shuffle.

*/

private Applications filterAndShuffle(Applications apps) {

if (apps != null) {
    if (isFetchingRemoteRegionRegistries()) {
        Map<String, Applications> remoteRegionVsApps = new ConcurrentHashMap<String, Applications>();
        apps.shuffleAndIndexInstances(remoteRegionVsApps, clientConfig, instanceRegionChecker);
        for (Applications applications : remoteRegionVsApps.values()) {
            applications.shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());
        }
        this.remoteRegionVsApps = remoteRegionVsApps;
    } else {
        apps.shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());
    }
}
return apps;

}

由此可以看出，是否需要过滤服务状态的配置是在clientConfig中，下一步寻找这个配置具体要怎么写。在类中可以看出，clientConfig对应的是EurekaClientConfig，但这是个接口，有两个实现

EurekaClientConfig

DefaultEurekaClientConfig

EurekaClientConfigBean

仅看名字以为是使用的DefaultEurekaClientConfig，找到对应的配置方法是：

/*

• (non-Javadoc)

*

• @see
• com.netflix.discovery.EurekaClientConfig#shouldFilterOnlyUpInstances()

*/

@Override

public boolean shouldFilterOnlyUpInstances() {

return configInstance.getBooleanProperty(
        namespace + SHOULD_FILTER_ONLY_UP_INSTANCES_KEY, true).get();

}

于是找到namespace以及对应的常量，组合起来的配置应该是：eureka.shouldFilterOnlyUpInstances，然而，加上配置后并没有什么用。仔细看看这是eureka包里的，而不是spring包里的，所以这应该是独立使用eureka的时候配置的方法，回过头来看，另外一个配置的实现（EurekaClientConfigBean）是spring cloud包里的，找找在哪里有用到？果然不出所料，在EurekaClientAutoConfiguration自动配置中有初始化此bean，并且EurekaClientConfigBean上有@ConfigurationProperties注解，其实也就是个properties。所以结论出来了，EurekaClientConfigBean实际是将properties配置和对EurekaClientConfig接口的实现放一起了。。。，这样就能找到配置不自动过滤状态为UP的服务的方法了

eureka.client.filterOnlyUpInstances=false

二. 动态更新EurekaClient的状态

首先，spring cloud官方文档中，介绍了一个endpoint（/service-registry/instance-status），提供了get和post方法，get方法用来获取节点状态，post用来修改节点状态

Service Registry Actuator Endpoint

A /service-registry actuator endpoint is provided by Commons. This endpoint relys on a Registration bean in the Spring Application Context. Calling /service-registry/instance-status via a GET will return the status of the Registration. A POST to the same endpoint with a String body will change the status of the current Registration to the new value. Please see the documentation of the ServiceRegistry implementation you are using for the allowed values for updating the status and the values retured for the status.

spring cloud eureka client默认是没有开启endpoint的，需要自己引入依赖

<dependency>

<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>

</dependency>

不过，引入依赖后会不止这一个endpoint，并且需要安全认证，当然也可以配置不需要安全认证：

management.security.enabled=false

为了不依赖actutator包，并且了解spring cloud具体是如何更新节点状态的，找到了endpoint中修改状态的具体实现：ServiceRegistryEndpoint。可以看出，主要是通过ServiceRegistry和Registration实现的，而这两个接口并不是actuator包里的，所以尝试自己实（拷）现（贝）一下。自己写一个controller，注入以上两个对象，然后将ServiceRegistryEndpoint中的获取和修改状态的方法复制粘贴，源码如下：

import javax.annotation.Resource;

import org.springframework.cloud.client.serviceregistry.ServiceRegistry;

import org.springframework.cloud.netflix.eureka.serviceregistry.EurekaRegistration;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestMethod;

import org.springframework.web.bind.annotation.RestController;

import com.itopener.framework.ResultMap;

@RestController

@RequestMapping("eureka-client")

public class ServiceRegistryController {

@Resource
private ServiceRegistry<EurekaRegistration> serviceRegistry;

@Resource
private EurekaRegistration registration;

@RequestMapping(value = "status", method = RequestMethod.GET)
public ResultMap getStatus(){
    return ResultMap.buildSuccess().put("status", serviceRegistry.getStatus(registration));
}

@RequestMapping(value = "status", method = RequestMethod.POST)
public ResultMap setStatus(String status){
    serviceRegistry.setStatus(registration, status);
    return ResultMap.buildSuccess();
}

}

需要注意的是

ServiceRegistry有Registration接口的实现类的泛型，如果不对应会注入失败，这样看具体是哪个实现类？ServiceRegistry只有一个实现类：EurekaServiceRegistry，所以结果就显而易见了

当然，这两个bean的初始化也会在自动配置类中（EurekaClientAutoConfiguration）

@Bean

public EurekaServiceRegistry eurekaServiceRegistry() {

return new EurekaServiceRegistry();

}

@Bean

@ConditionalOnBean(AutoServiceRegistrationProperties.class)

@ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true)

public EurekaRegistration eurekaRegistration(EurekaClient eurekaClient, CloudEurekaInstanceConfig instanceConfig,

ApplicationInfoManager applicationInfoManager) {
return EurekaRegistration.builder(instanceConfig)
    .with(applicationInfoManager)
    .with(eurekaClient)
    .with(healthCheckHandler)
    .build();

}

三. 动态管理spring cloud eureka服务

基于以上两点，就可以动态停用或启用eureka中注册的服务节点了。创建一个独立的web应用，与普通的服务一样注册到eureka中心，当然为了还是需要与其他服务有些不一样的配置

本节点不注册到eureka

eureka.client.register-with-eureka=false

可以从eureka拉取注册的服务

eureka.client.fetch-registry=true

不过滤服务节点的UP状态，即需要使用所有的服务节点

eureka.client.filterOnlyUpInstances=false

然后就可以使用EurekaClient获取注册中心的服务了

@Resource

private EurekaClient eurekaClient;

/**

• @description 获取服务数量和节点数量
• @author fuwei.deng
• @date 2017年7月21日 下午3:36:24
• @version 1.0.0
• @return

*/

@RequestMapping(value = "home", method = RequestMethod.GET)

public ResultMap home(){

List<Application> apps = eurekaClient.getApplications().getRegisteredApplications();
int appCount = apps.size();
int nodeCount = 0;
for(Application app : apps){
    nodeCount += app.getInstancesAsIsFromEureka().size();
}
return ResultMap.buildSuccess().put("appCount", appCount).put("nodeCount", nodeCount);

}

/**

• @description 获取所有服务节点
• @author fuwei.deng
• @date 2017年7月21日 下午3:36:38
• @version 1.0.0
• @return

*/

@RequestMapping(value = "apps", method = RequestMethod.GET)

public ResultMap apps(){

List<Application> apps = eurekaClient.getApplications().getRegisteredApplications();
Collections.sort(apps, new Comparator<Application>() {
    public int compare(Application l, Application r) {
        return l.getName().compareTo(r.getName());
    }
});
return ResultMap.buildSuccess().put("list", apps);

}

如果需要动态修改节点的状态，以达到停用和启用服务节点的目的，可以使用http调用对应节点的接口

@RequestMapping(value = "status/{appName}", method = RequestMethod.POST)

public ResultMap status(@PathVariable String appName, String instanceId, String status){

Application application = eurekaClient.getApplication(appName);
InstanceInfo instanceInfo = application.getByInstanceId(instanceId);
HttpUtil.post(instanceInfo.getHomePageUrl() + "eureka-client/status", "status=" + status);
return ResultMap.buildSuccess();

}

当然如果是使用服务节点的actuator endpoint接口，调用接口的地址不一样（还有安全认证，此处代码未涉及），需要注意的是，endpoint接收的参数是@RequestBody（并且使用的jackson转换，fastjson转换是会出现异常的）

@RequestMapping(value = "status/{appName}", method = RequestMethod.POST)

public ResultMap status(@PathVariable String appName, String instanceId, String status){

Application application = eurekaClient.getApplication(appName);
InstanceInfo instanceInfo = application.getByInstanceId(instanceId);
HttpUtil.post(instanceInfo.getHomePageUrl() + "service-registry/instance-status", status);
return ResultMap.buildSuccess();

}

由于eureka注册中心没有通知的功能，只能由节点自己发起刷新请求，所以修改状态后，需要等到相关节点下一次刷新后才会生效。节点刷新是通过定时任务实现的，源码在com.netflix.discovery.DiscoveryClient中，并且任务是在构造方法中初始化的，还不能自己手动触发，主要代码如下：

//任务调度器，私有属性

private final ScheduledExecutorService scheduler;

//刷新注册中心节点的线程池，私有属性

private final ThreadPoolExecutor cacheRefreshExecutor

//1224行，私有方法中，如果允许拉取注册中心的节点，则初始化调度任务，从源码中可以看出能配置任务执行的间隔时间

if (clientConfig.shouldFetchRegistry()) {

// registry cache refresh timer
int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
scheduler.schedule(
    new TimedSupervisorTask(
            "cacheRefresh",
            scheduler,
            cacheRefreshExecutor,
            registryFetchIntervalSeconds,
            TimeUnit.SECONDS,
            expBackOffBound,
            new CacheRefreshThread()
    ),
    registryFetchIntervalSeconds, TimeUnit.SECONDS);

}

配置在EurekaClientAutoConfiguration-->RefreshableEurekaClientConfiguration，使用DiscoveryClient的子类CloudEurekaClient实例化

@Bean(destroyMethod = "shutdown")

@ConditionalOnMissingBean(value = EurekaClient.class, search = SearchStrategy.CURRENT)

@org.springframework.cloud.context.config.annotation.RefreshScope

@Lazy

public EurekaClient eurekaClient(ApplicationInfoManager manager,

EurekaClientConfig config, EurekaInstanceConfig instance) {
manager.getInfo(); // force initialization
return new CloudEurekaClient(manager, config, this.optionalArgs,
        this.context);

}

CloudEurekaClient中有一个刷新的方法，发布一个心跳事件，但这个方法是protected，没法通过实例调用，并且依赖于心跳事件。应用节点默认刷新事件是60秒一次，时间也不算太长，所以动态停用节点后再60秒内生效，应该是在能接受的范围吧，并且这个时间还能配置

如果你也想在IT行业拿高薪，可以参加我们的训练营课程，选择最适合自己的课程学习，技术大牛亲授，7个月后，进入名企拿高薪。我们的课程内容有：Java工程化、高性能及分布式、高性能、深入浅出。高架构。性能调优、Spring，MyBatis，Netty源码分析和大数据等多个知识点。如果你想拿高薪的，想学习的，想就业前景好的，想跟别人竞争能取得优势的，想进阿里面试但担心面试不过的，你都可以来，群号为：575745314

注：加群要求

1、具有1-5工作经验的，面对目前流行的技术不知从何下手，需要突破技术瓶颈的可以加。

2、在公司待久了，过得很安逸，但跳槽时面试碰壁。需要在短时间内进修、跳槽拿高薪的可以加。

3、如果没有工作经验，但基础非常扎实，对 java 工作机制，常用设计思想，常用java开发框架掌握熟练的，可以加。

4、觉得自己很牛B，一般需求都能搞定。但是所学的知识点没有系统化，很难在技术领域继续突破的可以加。

5.阿里Java高级大牛直播讲解知识点，分享知识，多年工作经验的梳理和总结，带着大家全面、科学地建立自己的技术体系和技术认知！

6.小号或者小白之类加群一律不给过，谢谢。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

• Ray源码解析之调度部分
• CVPR2019 | 29篇目标检测相关论文汇总（部分含源码）
• Spring Boot--yml/properties配置文件以及部分源码
• [Spring-Analysis] v0.0.2 Spring Env 部分源码分析
• C# 中新增类型的命名空间只需部分与其他命名空间名称相同即可破坏源码兼容性
• 我入行以来java框架的几个阶段&论什么是真正的组件开发&项目中部分源码干货分享

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。