基于Redis实现Spring Cloud Gateway的动态管理

2023-02-27

数据微服务网关

引言：SpringCloudGateway是当前使用非常广泛的一种API网关。它本身能力并不能完全满足企业对网关的期望，人们希望它可以提供更多的服务治理能力。但SpringCloudGateway并不提供数据的动态管理，甚至修改个路由都需要重启。我们如何解决它这个短板，同时实现治理配置数据的高效动态

引言：

Spring Cloud Gateway是当前使用非常广泛的一种API网关。它本身能力并不能完全满足企业对网关的期望，人们希望它可以提供更多的服务治理能力。但Spring Cloud Gateway并不提供数据的动态管理，甚至修改个路由都需要重启。我们如何解决它这个短板，同时实现治理配置数据的高效动态管理呢?本文将带来我们网关与Redis组合的实践。

客户端会多次请求不同的微服务，增加了客户端的复杂性。
存在跨域请求，在一定场景下处理相对复杂。
认证复杂，每个服务都需要独立认证。
难以重构，随着项目的迭代，可能需要重新划分微服务。例如，可能将多个服务合并成一个或者将一个服务拆分成多个。如果客户端直接与微服务通信，那么重构将会很难实施。
某些微服务可能使用了防火墙 / 浏览器不友好的协议，直接访问会有一定的困难。

以上这些问题可以借助 API 网关解决。API 网关是介于客户端和服务器端之间的中间层，所有的外部请求都会先经过 API 网关这一层。也就是说，API 的实现方面更多的考虑。

使用 API 网关后的优点如下：

易于监控。可以在网关收集监控数据并将其推送到外部系统进行分析。
易于认证。可以在网关上进行认证，然后再将请求转发到后端的微服务，而无须在每个微服务中进行认证。
减少了客户端与各个微服务之间的交互次数。

Spring Cloud Gateway

Spring Cloud Gateway是Spring官方基于Spring 5.0，Spring Boot 2.0和Project Reactor等技术开发的网关，Spring Cloud Gateway旨在为微服务架构提供一种简单而有效的统一的API路由管理方式。

Spring Cloud Gateway作为Spring Cloud生态系中的网关，目标是替代Netflix ZUUL，其不仅提供统一的路由方式，并且基于Filter链的方式提供了网关基本的功能，例如：安全，监控/埋点，和限流等。

如图所示，SCG的架构看起来很简单。

首先，它内部包含了一个高性能的Netty Server，用来接收各类网络请求。请求进来之后，会根据配置的各个路由进行匹配并处理请求。每个路由都可以定义多个断言(Predicate)，用于路由匹配。

SCG默认提供了10多个内建的断言，可以基于请求的各个方面(请求头，路径，路径，时间，Cookie，http方法等)进行路由匹配。如果还不够，用户还可以自已扩展。

请求匹配到了合适的路由之后，就会按照路由中配置的各过滤器(filter)，按顺序对请求进行处理。Filter也基本上可以对请求的所有属性做处理，修改，添加或者除请求头，修改请求数据，修改返回的数据等，几乎无所不能。当然，修改请求也只是一方面的用途，认证，鉴权，记录日志等也都可以在网关中统一来做。

所有filter形成处理链，直到所有的filter处理完，才会交给最后面的 Netty Client，由它将处理过的请求发送至对应的微服务。

在请求发送至微服务之前，还可以定义它的负载均衡策略(LoadBalancerRule)，以决定请求至底发往微服务的哪个实例。

Filter 与 LoadBalancerRule 都支持自行扩展。

2.网关数据管理

实现一个适合自已的网关，对数据管理需要考虑哪些方面的东西呢?

（1）首先，我们要考虑一下，我们需要管理些什么数据。

SCG本身对数据管理的管理是很弱的。它没有提供数据的持久化方案，它所有的数据都来自初始化，来自它的配置文件(application.yml)。它本身虽然也对外提供了一些管理接口(Actuator API)能力不够，但能力不够，且这些修改都是暂时的，网关一停，数据就消失了。这就要求我们要用一套更完善的方案，把网关的这些数据管理起来，不能让它只能写在配置文件中，而要支持持久化，支持动态变更。再有就是我们对各微服务的治理数据。网关只用来做路由转发，那就太浪费了，统一认证，统一鉴权，访问日志记录，应用访问统计，黑白名单过滤，API订阅管理，流量限制，甚至数据格式转换，网络协议转换，都可以在网关中来做。而所有的这些能力，无不需要数据的支持。因此，这些服务的治理配置，也是网关需要管理的数据。

（2）数据有了，我们还得考虑怎么把它保存起来，不能网关一重启，所有数据就没了。

（3）还得再考虑一下数据的读取。网关对性能的要求是很高的，每次对过关的数据进行治理，都需要去读取这些配置信息。如果配置信息读取太消耗资源，无疑对网关是不利的。所以，我们还得考虑数据如何缓存，以提高数据的读取性能。

（4）单个网关，可以处理的请求量是有上限的。为了应对大的流量，我们可能会需要对网关做水平扩容。当多个网关实例共存时，如何保障对网关的修改，能快速同步到每个网关实例呢?数据变更通知也得考虑。

（5）最多，我们还得考虑一下方案的扩展，数据存储能不能改个地方，通知能不能换种方式?

综合考虑了这些方面之后，我们的网关的架构如下：

如图，以上就是我们网关的整体设计。方案设计要点如下：

网关对外提供治理数据管理接口, 微服务治理平台可通过这些接口, 将治理配置推送到网关
网关通过治理数据统一存储接口, 将治理配置数据保持至治理数据持久存储(这里我们默认为Redis)
Redis通过发布订阅能力, 将数据的变更通知到各网关实例
各网关实例收到通知后, 将数据从持久存储同步至内部高速缓存
内部缓存在网关启动时, 会自动从持久存储加载对应配置进入缓存. 同时它也支持清空, 以及按需加载
外部业务请求经过网关时, 对数据执行鉴权,处理转换, 以及灰度策略时,所需要治理配置,都从内部缓存中获取, 以提升性能
方案中, 外部持久存储(默认用的Redis, 可以换成Mysql, 文件, Appolo等), 以及数据变更通知(默认使用的是Redis的发布订阅, 可以换成Appolo通知, 消息队列, 定时扫描等), 都是可以扩展的

3.实现细节

动态路由管理

Spring Cloud Gateway作为所有请求流量的入口，在实际生产环境中为了保证高可靠和高可用，尽量避免重启, 需要实现Spring Cloud Gateway动态路由配置。实现动态路由其实很简单, 重点在于 RouteDefinitionRepository 这个接口. 这个接口继承自两个接口, 其中 RouteDefinitionLocator 是用来加载路由的. 它有很多实现类，其中的 PropertiesRouteDefinitionLocator 就用来实现从yml中加载路由。另一个 RouteDefinitionWriter 用来实现路由的添加与删除。通过查看spring cloud gateway的源码可以发现, 在 org.springframework.cloud.gateway.config.GatewayAutoConfiguration中这么一段：

@Bean 
@ConditionalOnMissingBean(RouteDefinitionRepository.class) 
public InMemoryRouteDefinitionRepository inMemoryRouteDefinitionRepository() { 
    return new InMemoryRouteDefinitionRepository(); 
} 
1.
2.
3.
4.
5.

可以看出, 网关中如果没有RouteDefinitionRepository的Bean, 就会采用InMemoryRouteDefinitionRepository做为实现。这个 InMemoryRouteDefinitionRepository有一个问题, 就是数据没有持久化, 网关重启之后,原来通过接口设置的路由就会丢失了。

这当然是不可接受的, 所以我们需要实现自已的 RouteDefinitionRepository, 来提供路由配置信息。如使用redis做为存储, 来实现路由的存储。实现请参考文章：https://dwz.cn/tsHfKwMe

除此以外, 每当路由更改之后, 还需要通知网关刷新路由。这需要发送 RefreshRoutesEvent 来通知网关。如下列示例：

@Component 
public class RouteDynamicService implements ApplicationEventPublisherAware { 
    private ApplicationEventPublisher publisher; 
 
    @Override 
    public void setApplicationEventPublisher(ApplicationEventPublisher publisher) { 
      this.publisher = publisher; 
    } 
 
    /** 
    * 刷新路由表 
    */ 
    public void refreshRoutes() { 
      publisher.publishEvent(new RefreshRoutesEvent(this)); 
    } 
} 
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.

刷新可以通过消息通知机制来触发, 当然, 也可以对外接供rest接口, 手动触发。

数据存储

如上述类图所示, IGovernDataRepository为治理数据统一存储接口。RedisGovernDataRepository为实现的它的抽像类, 它需要依赖两个, 一个是StringRedisTemplate，用来实现redis数据的存储。另一个为 RedisKeyGenerator, 用来为各治理对象生成对应的key。RedisGovernDataRepository下面则为各个治理数据存储的实现类。使用Redis做为持久存储时, 需要注意以下几点:

为对象生成key时, 建议为key添加一个命名空间(就是加一段有意义的前缀)
在redis中进行模糊搜索时, 提供给Redis的pattern, 不能是一个正则的通配, 它支持三种通配 *(多个), ?(单个)
如果数据量比较大, 不建议使用keys进行模糊查询, 应该使用scan方式

数据缓存

我们提供了内部缓存，它处于使用者与持久存储之间，缓存数据以提升性能。缓存的实现主要有如下几点：

实现了 InitializingBean 以实现在网关启动时, 自动加载数据
内部使用了ConcurrentHashMap, 保证写时的线程同步, 又保证了get时的高效(get整个过程不需要加锁)
从缓存中取数据时, 如果需要懒加载, 当从持久存储中加载不到数据时, 建议使用空数据, 或空集合占位, 避免每次都去持久存储中查询

代码示例如下：

/** 
  * 根据 appCode 获取流量策略 
  *  
  * @param appCode 
  * @return 
  */ 
public Set<ApplicationTrafficPolicy> getAppTrafficPolicies(String appCode) { 
  // 从缓存加载 
  Map<String, ApplicationTrafficPolicy> map = policyMap.get(appCode); 
  // 缓存中没有 
  if (map == null) { 
    // 尝试从持久存储中加载所有此网关的流量策略 
    Set<ApplicationTrafficPolicy> policies = trafficPolicyRepository.fuzzyQuery(); 
    // 持久存储中没有任何流量策略，占个位置，防止缓存重复去加载 
    if (policies == null || policies.size() == 0) { 
      map = new ConcurrentHashMap<>(); 
      policyMap.put(appCode, map); 
    } else { 
      // 持久存储中有流量策略，放入缓存 
      for (ApplicationTrafficPolicy policy : policies) { 
        setTrafficPolicy(policy); 
      } 
      // 重新从缓存中加载一次 
      map = policyMap.get(appCode); 
      // 如果还是没有，使用空 map 占位子 
      if (map == null) { 
        map = new ConcurrentHashMap<>(); 
        policyMap.put(appCode, map); 
      } 
    } 
  } 
  return map.values().stream().collect(Collectors.toSet()); 
} 
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.

事件通知

事件通知，这里我们使用的是redis的发布与订阅能力。Redis默认是不发送事件的，要让它发布事件，需要先修改它的配置文件redis.conf，添加一个配置：

notify-keyspace-events "K$g" 
1.

上面的配置将使得Redis中发生数据的添加，修改或删除时，发送set或del事件。

然后，我们需要配置一个RedisMessageListenerContainer，用来订阅我们感兴趣的事件。

@Bean 
RedisMessageListenerContainer container(MessageListenerAdapter listenerAdapter) { 
  String gtwReidsPattern = "__keyspace@*__:" + GTW + keyGenerator.getGatewayCode() + "]*"; 
  String cofRedisPattern = "__keyspace@*__:" + COF + cacheKey.getKeyNameSpace() + USER_NAME + "*"; 
  log.info("Add gateway redis message listener, patternTopic is {}", gtwReidsPattern); 
  log.info("Add coframe redis message listener, patternTopic is {}", cofRedisPattern); 
  RedisMessageListenerContainer container = new RedisMessageListenerContainer(); 
  container.setConnectionFactory(redisTemplate.getConnectionFactory()); 
  // PatternTopic 参考：http://redisdoc.com/topic/notification.html 
  container.addMessageListener(listenerAdapter, Arrays.asList(new PatternTopic(PatternUtil.fmt(gtwReidsPattern)), new PatternTopic(PatternUtil.fmt(cofRedisPattern)))); 
  return container; 
} 
当redis事件订阅好了之后, 每次其中我们关心的数据有变更, 都会发送set或del事件. 
我们需要定义一个 MessageListener, 来接收事件: 
@Service(value = RedisMessageListener.REDIS_LISTENER_NAME) 
public class RedisMessageListener implements MessageListener { 
  @Override 
  public void onMessage(Message message, byte[] pattern) { 
    String ops = new String(message.getBody()); 
    String channel = new String(message.getChannel()); 
    String key = channel.split(":")[1]; 
 
    if ("set".equals(ops)) { 
      String value = redisTemplate.opsForValue().get(key); 
      handleSet(key, value); 
    } else if ("del".equals(ops)) { 
      handleDel(key); 
    } 
  } 
  ... 
} 
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.