问题记录：SpringCloud 灰度环境路由 Gateway

版本:

springboot 2.0.3

springcloud  Finchley.SR3

 

首先创建一个filter用于过滤请求，判断灰度条件，给灰度请求添加灰度标识

[code]@Slf4j
public class EcsGrayFilter extends LoadBalancerClientFilter {

@Autowired
private SpringClientFactory clientFactory;

private EcsRibbonLoadBalancerClient loadBalancer;

public EcsGrayFilter(LoadBalancerClient loadBalancer) {
super(loadBalancer);
}

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
HttpHeaders httpHeaders = exchange.getRequest().getHeaders();
ServerHttpRequest request = exchange.getRequest();
//对头信息进行灰度判定 判断当前请求是否进入灰度环境
boolean isGray = false;
//此处灰度逻辑
if (isGray) {
//给请求头添加灰度标标
ServerHttpRequest newRequest = request.mutate().header(BaseConstant.ENVIRONMENT,BaseConstant.GRAY).build();
exchange = exchange.mutate().request(newRequest).build();
}
return super.filter(exchange,chain);
}

protected ServiceInstance choose(ServerWebExchange exchange) {
if (loadBalancer == null) {
loadBalancer = new EcsRibbonLoadBalancerClient(clientFactory);
}
return this.loadBalancer.choose(((URI)exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR)).getHost(), exchange.getRequest().getHeaders());
}
}

[code]将此过滤器提交给spring容器，即在启动类中增加如下方法：

[code]	@Bean
public EcsGrayFilter userLoadBalanceClientFilter(LoadBalancerClient client) {
return new EcsGrayFilter(client);
}

[code]以下是org.springframework.cloud.netflix.ribbon.RibbonLoadBalancerClient

重写choose() 和 getServer()方法

[code]public class EcsRibbonLoadBalancerClient extends RibbonLoadBalancerClient {

private SpringClientFactory clientFactory;

public EcsRibbonLoadBalancerClient(SpringClientFactory clientFactory) {
super(clientFactory);
}

private ServerIntrospector serverIntrospector(String serviceId) {
ServerIntrospector serverIntrospector = this.clientFactory.getInstance(serviceId,
ServerIntrospector.class);
if (serverIntrospector == null) {
serverIntrospector = new DefaultServerIntrospector();
}
return serverIntrospector;
}

protected Server getServer(ILoadBalancer loadBalancer,HttpHeaders headers) {
if (loadBalancer == null) {
return null;
}
//获取环境标记
String loadBalanceKey = headers.getFirst(BaseConstant.ENVIRONMENT);
return StringUtils.isEmpty(loadBalanceKey) ? loadBalancer.chooseServer(BaseConstant.DEFAULT):loadBalancer.chooseServer(loadBalanceKey);
}
}

将请头中的灰度标记传递给loadBalancer(负载均衡)

自定义一个负载均衡策略，这里负载均衡算法使用BestAvailableRule的算法，在此算法基础上增加灰度过滤

[code]public class EcsBestAvailableRule extends PredicateBasedRule{
private LoadBalancerStats loadBalancerStats;
private ILoadBalancer loadBalancer;
private AbstractServerPredicate predicate = new EcsDiscoveryEnabledPredicate();
private RoundRobinRule roundRobinRule; private String appName;
private String clientName;
private String eurekaUrls;
public void setEurekaUrls(String eurekaUrls) {
this.eurekaUrls = eurekaUrls;
}

@Override
public AbstractServerPredicate getPredicate() {
return predicate;
}

public Server choose(Object key) {
//此处自定义负载均衡策略
if (this.loadBalancerStats == null) {
return super.choose(key);
} else {
String keyStr = (String) key;
if (BaseConstant.GRAY.equals(keyStr)) {
//灰度使用默认负载
Server server = super.choose(key);
//如果灰度此请求没有对应的灰度服务则走在产服务,使用BestAvailableRule负载
if (server == null) {
server = bestAvailableRuleChoose(BaseConstant.DEFAULT);
}
return server;
} else {
//在产版本使用BestAvailableRule负载
return bestAvailableRuleChoose(key);
}
}
}

/**
* bestAvailableRule
* 此负载均衡策略：剔除熔断服务后在所有服务中选择连接数最小（负载最小的服务）
* @param key
* @return
*/
Server bestAvailableRuleChoose(Object key){
//获取所有服务
List<Server> serverListAll = this.getLoadBalancer().getAllServers();
//通过断言获取待负载均衡的所有服务
List<Server> serverList = this.predicate.getEligibleServers(serverListAll,key);
int minimalConcurrentConnections = 2147483647;
long currentTime = System.currentTimeMillis();
Server chosen = null;
Iterator var7 = serverList.iterator();
while(var7.hasNext()) {
Server server = (Server)var7.next();
ServerStats serverStats = this.loadBalancerStats.getSingleServerStat(server);
if (!serverStats.isCircuitBreakerTripped(currentTime)) {
int concurrentConnections = serverStats.getActiveRequestsCount(currentTime);
if (concurrentConnections < minimalConcurrentConnections) {
minimalConcurrentConnections = concurrentConnections;
chosen = server;
}
}
}

if (chosen == null) {
return super.choose(key);
} else {
return chosen;
}
}

public void setLoadBalancer(ILoadBalancer lb) {
super.setLoadBalancer(lb);
if (lb instanceof AbstractLoadBalancer) {
this.loadBalancerStats = ((AbstractLoadBalancer)lb).getLoadBalancerStats();
}

}
}

自定义一个断言

[code]public class EcsDiscoveryEnabledPredicate extends AbstractServerPredicate {

public boolean apply(PredicateKey input) {
Server server = input.getServer();
String routTo = (String) input.getLoadBalancerKey();
if (server instanceof DiscoveryEnabledServer) {
DiscoveryEnabledServer enabledServer = (DiscoveryEnabledServer) server;
InstanceInfo instanceInfo = enabledServer.getInstanceInfo();
//这里获取服务的环境属性，如果服务环境属性
String serverStatus = instanceInfo.getMetadata().get(BaseConstant.ENVIRONMENT);
//请求标记判断请求标记与环境是佛一致
switch (routTo) {
case BaseConstant.GRAY:
return BaseConstant.GRAY.equals(serverStatus);
case BaseConstant.DEFAULT:
if (BaseConstant.GRAY.equals(serverStatus) || StringUtils.isBlank(serverStatus)) {
return false;
}
return true;
default:
return false;
}
}
return false;
}

public List<Server> getEligibleServers(List<Server> servers, Object loadBalancerKey) {
if (loadBalancerKey == null) {
return ImmutableList.copyOf(Iterables.filter(servers, this.getServerOnlyPredicate()));
} else {
List<Server> results = Lists.newArrayList();
Iterator var4 = servers.iterator();

while(var4.hasNext()) {
Server server = (Server)var4.next();
//这里将所有服务进行过滤，排除掉不符合条件的服务
if (this.apply(new PredicateKey(loadBalancerKey, server))) {
results.add(server);
}
}

return results;
}
}
}

对需要进行灰度路由的服务配置自定义的负载均衡策略

[code]yourServiceName:
ribbon:
#BestAvailableRule负载均衡策略
NFLoadBalancerRuleClassName: com.xxx.EcsBestAvailableRule

这里yourServiceName是注册到注册中心的服务名，属性NFLoadBalancerRuleClassName的值为自定义负载均衡类的全路径。

需要进行灰度路由微服务配置文件中增加环境标志，这里是在bootstrap.yml中进行配置，将环境标记增加到metadata-map中（前一步断言处理中获取该属性），profile通过maven控制

[code]eureka:
instance:
#eureka客户端需要多长时间发送心跳给eureka服务器，表明他仍然或者，默认30秒
lease-renewal-interval-in-seconds: 30
#eureka服务器在接受到实力的最后一次发出的心跳后，需要等待多久才可以将此实力删除
lease-expiration-duration-in-seconds: 90
#使用ip注册
prefer-ip-address: true
#instance-id 使用IP:PORT
instance-id: ${spring.cloud.client.ip-address}:${server.port}
#在metadata-map中增加环境标记
metadata-map:
environment: ${maven.profiles.active}
client:
#表示eureka client间隔多久去拉取服务器注册信息,默认为30秒
registry-fetch-interval-seconds: 30
service-url:
defaultZone: ${maven.eureka.urls}

其中BaseConstant为自定义的静态变量类

总结：

1.在gateway中创建全局过滤器，对请求头进行灰度判断，对灰度请求打标。

2.重写客户端负载均衡方法（RibbonLoadBalancerClient中的choose()），将灰度标记进行传递。

3.重写负载均衡算法，选择最终要调用的服务。

4.重写断言，剔除不符合条件的服务(这里指的不是的服务灰度环境)。

5.配置自定义负载均衡策略

6.被调用的服务分环境部署，及启动时将环境标记加入服务实例的matedate-map

大概调用流程（省略部分）filter()->RibbonLoadBalancerClient.choose()->BaseLoadBalancer.chooseServer()->EcsBestAvailableRule.choose()->EcsDiscoveryEnabledPredicate.getEligibleServers()->EcsDiscoveryEnabledPredicate.apply()

 

• 点赞
• 收藏
• 分享
• 文章举报

qq_36707579 发布了4 篇原创文章 · 获赞 0 · 访问量 75 私信 关注