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

本文涉及的产品
云原生网关 MSE Higress,422元/月
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
注册配置 MSE Nacos/ZooKeeper,118元/月
简介:

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 <em>applications</em> 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秒内生效,应该是在能接受的范围吧,并且这个时间还能配置

目录
相关文章
|
2月前
|
Dubbo Java 应用服务中间件
Spring Cloud Dubbo:微服务通信的高效解决方案
【10月更文挑战第15天】随着信息技术的发展,微服务架构成为企业应用开发的主流。Spring Cloud Dubbo结合了Dubbo的高性能RPC和Spring Cloud的生态系统,提供高效、稳定的微服务通信解决方案。它支持多种通信协议,具备服务注册与发现、负载均衡及容错机制,简化了服务调用的复杂性,使开发者能更专注于业务逻辑的实现。
68 2
|
2天前
|
Java Nacos Sentinel
Spring Cloud Alibaba:一站式微服务解决方案
Spring Cloud Alibaba(简称SCA) 是一个基于 Spring Cloud 构建的开源微服务框架,专为解决分布式系统中的服务治理、配置管理、服务发现、消息总线等问题而设计。
68 12
Spring Cloud Alibaba:一站式微服务解决方案
|
3月前
|
Java 对象存储 开发者
解析Spring Cloud与Netflix OSS:微服务架构中的左右手如何协同作战
Spring Cloud与Netflix OSS不仅是现代微服务架构中不可或缺的一部分,它们还通过不断的技术创新和社区贡献推动了整个行业的发展。无论是对于初创企业还是大型组织来说,掌握并合理运用这两套工具,都能极大地提升软件系统的灵活性、可扩展性以及整体性能。随着云计算和容器化技术的进一步普及,Spring Cloud与Netflix OSS将继续引领微服务技术的发展潮流。
64 0
|
10天前
|
Prometheus 监控 Java
如何全面监控所有的 Spring Boot 微服务
如何全面监控所有的 Spring Boot 微服务
24 3
|
2月前
|
Java 数据库 数据安全/隐私保护
Spring 微服务提示:使用环境变量抽象数据库主机名
Spring 微服务提示:使用环境变量抽象数据库主机名
45 1
|
2月前
|
监控 Java 对象存储
监控与追踪:如何利用Spring Cloud Sleuth和Netflix OSS工具进行微服务调试
监控与追踪:如何利用Spring Cloud Sleuth和Netflix OSS工具进行微服务调试
46 1
|
2月前
|
安全 Java 对象存储
安全性考量:Spring Security与Netflix OSS在微服务安全中的作用
安全性考量:Spring Security与Netflix OSS在微服务安全中的作用
43 1
|
3月前
|
负载均衡 Java 网络架构
实现微服务网关:Zuul与Spring Cloud Gateway的比较分析
实现微服务网关:Zuul与Spring Cloud Gateway的比较分析
138 5
|
2月前
|
负载均衡 算法 Nacos
SpringCloud 微服务nacos和eureka
SpringCloud 微服务nacos和eureka
67 0
|
3月前
|
Java 对象存储 开发者
微服务世界的双雄争霸:Spring Cloud与Netflix OSS——谁将引领下一次企业级应用变革的风暴?
Spring Cloud与Netflix OSS是微服务架构的核心组件集,分别以其与Spring Boot的紧密集成及为大规模分布式系统设计的特性,在Java开发社区中广受青睐。前者通过Eureka提供服务发现机制,简化服务注册与定位;后者借助Hystrix增强系统弹性和可靠性,避免雪崩效应。此外,二者还包含负载均衡(Ribbon)、声明式HTTP客户端(Feign)及API网关(Zuul)等功能,共同构建强大微服务体系,助力开发者聚焦业务逻辑,提升系统灵活性与性能。
50 0