基于springcloud gateway + nacos实现灰度发布(reactive版)
什么是灰度发布?
灰度发布(又名金丝雀发布)是指在黑与白之间,能够平滑过渡的一种发布方式。在其上可以进行A/B testing,即让一部分用户继续用产品特性A,一部分用户开始用产品特性B,如果用户对B没有什么反对意见,那么逐步扩大范围,把所有用户都迁移到B上面来。灰度发布可以保证整体系统的稳定,在初始灰度的时候就可以发现、调整问题,以保证其影响度。
本文以springcloud gateway + nacos来演示如何实现灰度发布,如果对springcloud gateway和nacos还不熟悉的朋友,可以先阅读如下文章,然后再阅读本文。
实现的整体思路:
- 编写带权重的灰度路由
- 编写自定义filter
- nacos服务配置需要灰度发布的服务的元数据信息以及权重
- 灰度路由从nacos服务拉取元数据信息以及权重,然后根据权重算法,返回符合要求的服务实例给自定义的filter
- 网关配置文件配置需要灰度路由的服务(因为本文代码没有网关实现动态路由,不然灰度路由可以配置在配置中心,从配置中心拉取)
- filter通过责任链模式,把服务实例透传给其他filter比如NettyRoutingFilter
下边进入实战
正文
1、所使用的开发版本
<jdk.version>1.8</jdk.version> <!-- spring cloud --> <spring-cloud.version>Hoxton.SR3</spring-cloud.version> <spring-boot.version>2.2.5.RELEASE</spring-boot.version> <spring-cloud-alibaba.version>2.2.1.RELEASE</spring-cloud-alibaba.version>
2、pom.xml引入
<dependencies> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-gateway</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency> <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> <exclusions> <exclusion> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-ribbon</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-loadbalancer</artifactId> </dependency> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> </dependency> </dependencies>
ps:nacos的jar注意排除ribbon依赖,不然loadbalancer无法生效
3、编写权重路由
public class GrayLoadBalancer implements ReactorServiceInstanceLoadBalancer { private static final Log log = LogFactory.getLog(GrayLoadBalancer.class); private ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider; private String serviceId; public GrayLoadBalancer(ObjectProvider<ServiceInstanceListSupplier> serviceInstanceListSupplierProvider, String serviceId) { this.serviceId = serviceId; this.serviceInstanceListSupplierProvider = serviceInstanceListSupplierProvider; } @Override public Mono<Response<ServiceInstance>> choose(Request request) { HttpHeaders headers = (HttpHeaders) request.getContext(); if (this.serviceInstanceListSupplierProvider != null) { ServiceInstanceListSupplier supplier = (ServiceInstanceListSupplier)this.serviceInstanceListSupplierProvider.getIfAvailable(NoopServiceInstanceListSupplier::new); return ((Flux)supplier.get()).next().map(list->getInstanceResponse((List<ServiceInstance>)list,headers)); } return null; } private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> instances,HttpHeaders headers) { if (instances.isEmpty()) { return getServiceInstanceEmptyResponse(); } else { return getServiceInstanceResponseWithWeight(instances); } } /** * 根据版本进行分发 * @param instances * @param headers * @return */ private Response<ServiceInstance> getServiceInstanceResponseByVersion(List<ServiceInstance> instances, HttpHeaders headers) { String versionNo = headers.getFirst("version"); System.out.println(versionNo); Map<String,String> versionMap = new HashMap<>(); versionMap.put("version",versionNo); final Set<Map.Entry<String,String>> attributes = Collections.unmodifiableSet(versionMap.entrySet()); ServiceInstance serviceInstance = null; for (ServiceInstance instance : instances) { Map<String,String> metadata = instance.getMetadata(); if(metadata.entrySet().containsAll(attributes)){ serviceInstance = instance; break; } } if(ObjectUtils.isEmpty(serviceInstance)){ return getServiceInstanceEmptyResponse(); } return new DefaultResponse(serviceInstance); } /** * * 根据在nacos中配置的权重值,进行分发 * @param instances * * @return */ private Response<ServiceInstance> getServiceInstanceResponseWithWeight(List<ServiceInstance> instances) { Map<ServiceInstance,Integer> weightMap = new HashMap<>(); for (ServiceInstance instance : instances) { Map<String,String> metadata = instance.getMetadata(); System.out.println(metadata.get("version")+"-->weight:"+metadata.get("weight")); if(metadata.containsKey("weight")){ weightMap.put(instance,Integer.valueOf(metadata.get("weight"))); } } WeightMeta<ServiceInstance> weightMeta = WeightRandomUtils.buildWeightMeta(weightMap); if(ObjectUtils.isEmpty(weightMeta)){ return getServiceInstanceEmptyResponse(); } ServiceInstance serviceInstance = weightMeta.random(); if(ObjectUtils.isEmpty(serviceInstance)){ return getServiceInstanceEmptyResponse(); } System.out.println(serviceInstance.getMetadata().get("version")); return new DefaultResponse(serviceInstance); } private Response<ServiceInstance> getServiceInstanceEmptyResponse() { log.warn("No servers available for service: " + this.serviceId); return new EmptyResponse(); }
4、自定义filter
public class GrayReactiveLoadBalancerClientFilter implements GlobalFilter, Ordered { private static final Log log = LogFactory.getLog(ReactiveLoadBalancerClientFilter.class); private static final int LOAD_BALANCER_CLIENT_FILTER_ORDER = 10150; private final LoadBalancerClientFactory clientFactory; private LoadBalancerProperties properties; public GrayReactiveLoadBalancerClientFilter(LoadBalancerClientFactory clientFactory, LoadBalancerProperties properties) { this.clientFactory = clientFactory; this.properties = properties; } @Override public int getOrder() { return LOAD_BALANCER_CLIENT_FILTER_ORDER; } @Override public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { URI url = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR); String schemePrefix = (String)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_SCHEME_PREFIX_ATTR); if (url != null && ("grayLb".equals(url.getScheme()) || "grayLb".equals(schemePrefix))) { ServerWebExchangeUtils.addOriginalRequestUrl(exchange, url); if (log.isTraceEnabled()) { log.trace(ReactiveLoadBalancerClientFilter.class.getSimpleName() + " url before: " + url); } return this.choose(exchange).doOnNext((response) -> { if (!response.hasServer()) { throw NotFoundException.create(this.properties.isUse404(), "Unable to find instance for " + url.getHost()); } else { URI uri = exchange.getRequest().getURI(); String overrideScheme = null; if (schemePrefix != null) { overrideScheme = url.getScheme(); } DelegatingServiceInstance serviceInstance = new DelegatingServiceInstance((ServiceInstance)response.getServer(), overrideScheme); URI requestUrl = this.reconstructURI(serviceInstance, uri); if (log.isTraceEnabled()) { log.trace("LoadBalancerClientFilter url chosen: " + requestUrl); } exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR, requestUrl); } }).then(chain.filter(exchange)); } else { return chain.filter(exchange); } } protected URI reconstructURI(ServiceInstance serviceInstance, URI original) { return LoadBalancerUriTools.reconstructURI(serviceInstance, original); } private Mono<Response<ServiceInstance>> choose(ServerWebExchange exchange) { URI uri = (URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR); GrayLoadBalancer loadBalancer = new GrayLoadBalancer(clientFactory.getLazyProvider(uri.getHost(), ServiceInstanceListSupplier.class), uri.getHost()); if (loadBalancer == null) { throw new NotFoundException("No loadbalancer available for " + uri.getHost()); } else { return loadBalancer.choose(this.createRequest(exchange)); } } private Request createRequest(ServerWebExchange exchange) { HttpHeaders headers = exchange.getRequest().getHeaders(); Request<HttpHeaders> request = new DefaultRequest<>(headers); return request; } }
5、配置自定义filter给spring管理
@Configuration public class GrayGatewayReactiveLoadBalancerClientAutoConfiguration { public GrayGatewayReactiveLoadBalancerClientAutoConfiguration() { } @Bean @ConditionalOnMissingBean({GrayReactiveLoadBalancerClientFilter.class}) public GrayReactiveLoadBalancerClientFilter grayReactiveLoadBalancerClientFilter(LoadBalancerClientFactory clientFactory, LoadBalancerProperties properties) { return new GrayReactiveLoadBalancerClientFilter(clientFactory, properties); } }
6、编写网关application.yml配置
server: port: 9082 # 配置输出日志 logging: level: org.springframework.cloud.gateway: TRACE org.springframework.http.server.reactive: DEBUG org.springframework.web.reactive: DEBUG reactor.ipc.netty: DEBUG #开启端点 management: endpoints: web: exposure: include: '*' spring: application: name: gateway-reactor-gray cloud: nacos: discovery: server-addr: localhost:8848 gateway: discovery: locator: enabled: true lower-case-service-id: true routes: - id: hello-consumer uri: grayLb://hello-consumer predicates: - Path=/hello/**
uri中的grayLb配置,代表该服务需要进行灰度发布
7、在注册中心nacos配置灰度发布的服务版本以及权重值
weight代表权重,version代表版本
总结
上述就是实现灰度发布的过程,实现灰度发布的方法有很多种,文章中只是提供一种思路。虽然springcloud官方推荐使用loadbalancer来代替ribbon。因为ribbon是阻塞的,但从官方的loadbalancer的负载均衡算法来看,目前loadbalancer默认只支持轮询算法,要其他算法得自己扩展实现,而ribbon默认支持7种算法,用默认的算法基本上就可以满足我们的需求了。其次ribbon支持懒加载处理,超时以及重试与断路器hystrix集成等配置,loadbalancer目前就支持重试。所以如果正式环境要自己实现灰度发布,可以考虑对ribbon进行扩展。本文的实现只是作为一种扩展补充,毕竟springcloud推荐loadbalancer,索性就写个demo实现下。
最后灰度发布的实现,业内也有开源的实现--Discovery,感兴趣的朋友可以通过如下链接进行查看
demo链接
本文由博客群发一文多发等运营工具平台 OpenWrite 发布
- 基于Redis实现Spring Cloud Gateway的动态管理
- 基于Spring Cloud Gateway + STOMP 实现WebSocket负载均衡
- Spring Cloud Alibaba练习Demo(一):使用Nacos实现服务注册与发现
- Spring Cloud Alibaba基础教程:使用Nacos实现服务注册与发现
- spring cloud gateway 实现熔断降级攻略(原来gateway还集成Netty)
- 微服务电商系统从设计到实现基于springcloud
- spring cloud gateway整合sentinel实现网关限流
- Spring Cloud Alibaba基础教程:使用Nacos实现服务注册与发现
- spring cloud gateway 实现接口限流
- IDEA基于Spring Cloud Netflix(2.1.0RC3)的Spring Cloud Eureka来实现服务治理的微服务架构搭建以及和SSM框架的整合——实战教程
- 基于spring-cloud实现eureka注册服务小案例
- Spring Cloud Gateway重试机制的实现
- Spring Cloud Alibaba从入门到放弃-Nacos集成Sentinel实现限流降级熔断
- Spring Cloud Alibaba基础教程之Nacos实现服务注册与发现
- Spring Cloud下基于OAUTH2认证授权的实现示例
- Spring Cloud Gateway的动态路由怎样做?集成Nacos实现很简单
- 基于Spring cloud ribbon实现多版本控制
- 基于spring-cloud相关技术整合,实现接口调用、服务容错、动态路由配置等
- 求基于 SpringCloud 微服务架构下广告系统设计与实现
- 详解Spring Cloud Gateway基于服务发现的默认路由规则