MNbErvItc 发表于 2025-2-14 21:08:30

SpringCloud自定义loadbalancer实现标签路由

一、背景

  最近前端反应开发环境有时候调接口会很慢,原因是有开发图方便将本地服务注册到开发环境,请求路由到开发本地导致,
为了解决该问题想到可以通过标签路由的方式避免该问题,实现前端联调和开发自测互不干扰。
  该方案除了用于本地调试,还可以用于用户灰度发布。
二、实现方案

  关于负载均衡,低版本的SpringCloud用的是Spring Cloud Ribbon,高版本用Spring Cloud LoadBalancer替代了,
Ribbon可以通过实现IRlue接口实现,这里只介绍高版本的实现方案。
实现方案:

[*]idea在环境变量中设置tag,本地服务启动时读取环境变量将tag注册到nacos的元数据
[*]重写网关的负载均衡算法,从请求头中获取到的request-tag和服务实例的元数据进行匹配,如果匹配到则返回对应的
服务实例,否则提示服务未找到。
三、编码实现

3.1 order服务

新建一个SpringCloud服务order-service,注册元数据很简单,只需要排除掉NacosDiscoveryClientConfiguration,再写一个自己的NacosDiscoveryClientConfiguration配置类即可。
创建MyNacosDiscoveryClientConfiguration
/** * @Author: Ship * @Description: * @Date: Created in 2025/2/12 */@Configuration(      proxyBeanMethods = false)@ConditionalOnDiscoveryEnabled@ConditionalOnBlockingDiscoveryEnabled@ConditionalOnNacosDiscoveryEnabled@AutoConfigureBefore({SimpleDiscoveryClientAutoConfiguration.class, CommonsClientAutoConfiguration.class})@AutoConfigureAfter({NacosDiscoveryAutoConfiguration.class})public class MyNacosDiscoveryClientConfiguration {    @Bean    public DiscoveryClient nacosDiscoveryClient(NacosServiceDiscovery nacosServiceDiscovery) {      return new NacosDiscoveryClient(nacosServiceDiscovery);    }    @Bean    @ConditionalOnProperty(            value = {"spring.cloud.nacos.discovery.watch.enabled"},            matchIfMissing = true    )    public NacosWatch nacosWatch(NacosServiceManager nacosServiceManager, NacosDiscoveryProperties nacosDiscoveryProperties,                                 ObjectProvider<ThreadPoolTaskScheduler> taskExecutorObjectProvider, Environment environment) {      // 环境变量读取标签      String tag = environment.getProperty("tag");      nacosDiscoveryProperties.getMetadata().put("request-tag", tag);      return new NacosWatch(nacosServiceManager, nacosDiscoveryProperties, taskExecutorObjectProvider);    }}这里代码基本与NacosDiscoveryClientConfiguration一致,只是加上了设置元数据的逻辑。
@SpringBootApplication(exclude = NacosDiscoveryClientConfiguration.class)public class OrderApplication {    public static void main(String[] args) {      SpringApplication.run(OrderApplication.class, args);    }}启动类上需要排除默认的NacosDiscoveryClientConfiguration,不然启动会报bean重复注册的错误,或者配置添加spring.main.allow-bean-definition-overriding=true允许重复注册也行。
写一个测试接口,方便后面测试
/** * @Author: Ship * @Description: * @Date: Created in 2025/2/12 */@RequestMapping("test")@RestControllerpublic class TestController {    @GetMapping("")    public String sayHello(){      return "hello";    }}3.2 gateway服务

新建一个网关服务,pom文件如下:
<properties>      <java.version>1.8</java.version>      <spring-cloud.version>2020.0.3</spring-cloud.version>      <spring-cloud-alibaba.version>2021.1</spring-cloud-alibaba.version>      <spring-boot.version>2.5.1</spring-boot.version>      <maven-compiler-plugin.version>3.1</maven-compiler-plugin.version>    </properties>    <dependencies>      <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-web</artifactId>      </dependency>      <dependency>            <groupId>org.springframework.cloud</groupId>            <artifactId>spring-cloud-starter-loadbalancer</artifactId>      </dependency>      <dependency>            <groupId>org.springframework.cloud</groupId>            <artifactId>spring-cloud-starter-openfeign</artifactId>      </dependency>      <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-test</artifactId>            <version>${spring-boot.version}</version>            <scope>test</scope>      </dependency>      <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter</artifactId>            <version>${spring-boot.version}</version>      </dependency>      <dependency>            <groupId>com.alibaba.cloud</groupId>            <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>            <version>${spring-cloud-alibaba.version}</version>      </dependency>      <dependency>            <groupId>com.alibaba.cloud</groupId>            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>            <version>${spring-cloud-alibaba.version}</version>      </dependency>    </dependencies>    <dependencyManagement>      <dependencies>            <dependency>                <groupId>org.springframework.cloud</groupId>                <artifactId>spring-cloud-dependencies</artifactId>                <version>${spring-cloud.version}</version>                <type>pom</type>                <scope>import</scope>            </dependency>            <dependency>                <groupId>org.springframework.boot</groupId>                <artifactId>spring-boot-dependencies</artifactId>                <version>${spring-boot.version}</version>                <type>pom</type>                <scope>import</scope>            </dependency>            <dependency>                <groupId>org.springframework.cloud</groupId>                <artifactId>spring-cloud-starter-bootstrap</artifactId>            </dependency>      </dependencies>    </dependencyManagement>    <build>      <plugins>            <plugin>                <artifactId>maven-compiler-plugin</artifactId>                <version>${maven-compiler-plugin.version}</version>                <configuration>                  <source>${java.version}</source>                  <target>${java.version}</target>                </configuration>            </plugin>            <plugin>                <groupId>org.springframework.boot</groupId>                <artifactId>spring-boot-maven-plugin</artifactId>            </plugin>      </plugins>    </build>Spring-Cloud-loadBalancer默认使用轮询的算法,即org.springframework.cloud.loadbalancer.core.RoundRobinLoadBalancer类实现,因此可以参考RoundRobinLoadBalancer实现一个TagLoadBalancer,代码如下:
/** * @Author: Ship * @Description: * @Date: Created in 2025/2/12 */public class TagLoadBalancer implements ReactorServiceInstanceLoadBalancer {    private static final String TAG_HEADER = "request-tag";    private static final Log log = LogFactory.getLog(TagLoadBalancer.class);    final AtomicInteger position;    final String serviceId;    ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider;    public TagLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider, String serviceId) {      this(serviceInstanceListSupplierProvider, serviceId, (new Random()).nextInt(1000));    }    public TagLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider, String serviceId, int seedPosition) {      this.serviceId = serviceId;      this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider;      this.position = new AtomicInteger(seedPosition);    }    @Override    public Mono<Response<ServiceInstance>> choose(Request request) {      ServiceInstanceListSupplier supplier = (ServiceInstanceListSupplier) this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new);      return supplier.get(request).next().map((serviceInstances) -> {            return this.processInstanceResponse(supplier, serviceInstances, request);      });    }    private Response<ServiceInstance> processInstanceResponse(ServiceInstanceListSupplier supplier, List<ServiceInstance> serviceInstances, Request request) {      Response<ServiceInstance> serviceInstanceResponse = this.getInstanceResponse(serviceInstances, request);      if (supplier instanceof SelectedInstanceCallback && serviceInstanceResponse.hasServer()) {            ((SelectedInstanceCallback) supplier).selectedServiceInstance((ServiceInstance) serviceInstanceResponse.getServer());      }      return serviceInstanceResponse;    }    private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances, Request request) {      if (instances.isEmpty()) {            if (log.isWarnEnabled()) {                log.warn("No servers available for service: " + this.serviceId);            }            return new EmptyResponse();      }      if (request instanceof DefaultRequest) {            DefaultRequest<RequestDataContext> defaultRequest = (DefaultRequest) request;            // 上下文获取请求头            HttpHeaders headers = defaultRequest.getContext().getClientRequest().getHeaders();            List<String> list = headers.get(TAG_HEADER);            if (!CollectionUtils.isEmpty(list)) {                String requestTag = list.get(0);                for (ServiceInstance instance : instances) {                  String str = instance.getMetadata().getOrDefault(TAG_HEADER, "");                  if (requestTag.equals(str)) {                        return new DefaultResponse(instance);                  }                }                log.error(String.format("No servers available for service:%s,tag:%s ", this.serviceId, requestTag));                return new EmptyResponse();            }      }      int pos = Math.abs(this.position.incrementAndGet());      ServiceInstance instance = instances.get(pos % instances.size());      return new DefaultResponse(instance);    }}这里需要实现ReactorServiceInstanceLoadBalancer接口,如果请求头带有标签则根据标签路由,否则使用默认的轮询算法。
还要把TagLoadBalancer用起来,所以需要定义一个配置类TagLoadBalancerConfig,并通过@LoadBalancerClients注解添加默认配置,代码如下:
/** * @Author: Ship * @Description: * @Date: Created in 2025/2/12 */public class TagLoadBalancerConfig {    @Bean    public ReactorLoadBalancer reactorTagLoadBalancer(Environment environment, LoadBalancerClientFactory loadBalancerClientFactory) {      String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);      return new TagLoadBalancer(loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class), name);    }}@LoadBalancerClients(defaultConfiguration = {TagLoadBalancerConfig.class})@SpringBootApplicationpublic class GatewayApplication {    public static void main(String[] args) {      SpringApplication.run(GatewayApplication.class, args);    }}最后在application.yml文件添加网关路由配置
spring:application:    name: gatewaycloud:    nacos:      config:      server-addr: 127.0.0.1:8848      namespace: dev      group: DEFAULT_GROUP      discovery:      server-addr: 127.0.0.1:8848      namespace: dev    gateway:      routes:      - id: order-service          uri: lb://order-service          predicates:            - Path=/order/**          filters:            - StripPrefix=1server:port: 90003.3 代码测试


[*]本地启动nacos后启动order(注意需要在idea设置环境变量tag=ship)和gateway服务,可以看到order服务已经成功注册了元数据

[*]然后用Postman请求网关http://localhost:9000/order/test

可以看到请求成功路由到了order服务,说明根据tag路由成功了。

[*]去掉环境变量tag后重新启动Order服务,再次请求响应报文如下:
{    "timestamp": "2025-02-14T12:10:44.294+00:00",    "path": "/order/test",    "status": 503,    "error": "Service Unavailable",    "requestId": "41651188-4"}
说明根据requst-tag找不到对应的服务实例,代码逻辑生效了。
四、总结

  聪明的人已经发现了,本文只实现了网关路由到下游服务这部分的标签路由,下游服务A调服务B的标签路由并未实现,其实现方案也不难,只需要通过上下文传递+feign拦截器就可以做到全链路的标签路由,有兴趣的可以自己试试。

  本文代码已上传github,顺便推广下前段时间写的idea插件CodeFaster(快速生成常用流操作的代码,Marketplace搜索下载即可体验)😂。
页: [1]
查看完整版本: SpringCloud自定义loadbalancer实现标签路由