SpringCloud Gateway 动态路由

@Beanpublic RoutePredicateHandlerMapping routePredicateHandlerMapping(    FilteringWebHandler webHandler, RouteLocator routeLocator,    GlobalCorsProperties globalCorsProperties, Environment environment) {  return new RoutePredicateHandlerMapping(webHandler, routeLocator,      globalCorsProperties, environment);}

@Overrideprotected Mono<?> getHandlerInternal(ServerWebExchange exchange) {//... 忽略之上代码​  // lookupRoute 用于查找路由  return lookupRoute(exchange)      // 将查找到的路由记录到 ServerWebExchange 上下文中，然后，返回 FilteringWebHandler      .flatMap((Function<Route, Mono<?>>) r -> {        exchange.getAttributes().remove(GATEWAY_PREDICATE_ROUTE_ATTR);        if (logger.isDebugEnabled()) {          logger.debug(              "Mapping [" + getExchangeDesc(exchange) + "] to " + r);        }        // 后续会从Attributes获取到路由对象，进而获取路由过滤器，执行过滤器等列操作        exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, r);        return Mono.just(webHandler);​//... 忽略后续代码

 protected Mono<Route> lookupRoute(ServerWebExchange exchange) {   // this.routeLocator.getRoutes() 该方法是重点，后续需要继续分析    // RouteLocator 的实现类 是如何 getRoutes()   return this.routeLocator.getRoutes()       .concatMap(route -> Mono.just(route).filterWhen(r -> {         exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR, r.getId());         // 根据请求、当前路由的断言，判断当前当前路由断言是否命中，非本文重点，不展开         return r.getPredicate().apply(exchange);       })//... 忽略后续代码

// 转化方法 @Overridepublic Flux<Route> getRoutes() {  // 注意这里的 getRouteDefinitions()   Flux<Route> routes = this.routeDefinitionLocator.getRouteDefinitions()      .map(this::convertToRoute);  // ... 下略

@Beanpublic RouteLocator routeDefinitionRouteLocator(GatewayProperties properties,  // 注入所有过滤器工厂(不含全局过滤器)，用于根据RouteDefinition组装Route  List<GatewayFilterFactory> gatewayFilters,  // 注入所有断言工厂，用于根据RouteDefinition组装Route  List<RoutePredicateFactory> predicates,  // 注入RouteDefinition的加载器，这里是下文重点  RouteDefinitionLocator routeDefinitionLocator,  ConfigurationService configurationService) {  return new RouteDefinitionRouteLocator(routeDefinitionLocator, predicates,      gatewayFilters, properties, configurationService);}​@Bean@Primary@ConditionalOnMissingBean(name = "cachedCompositeRouteLocator")public RouteLocator cachedCompositeRouteLocator(List<RouteLocator> routeLocators) {  return new CachingRouteLocator(      new CompositeRouteLocator(Flux.fromIterable(routeLocators)));}

public RouteDefinitionRouteLocator(RouteDefinitionLocator routeDefinitionLocator,    List<RoutePredicateFactory> predicates,    List<GatewayFilterFactory> gatewayFilterFactories,    GatewayProperties gatewayProperties,    ConfigurationService configurationService) {  this.routeDefinitionLocator = routeDefinitionLocator;  this.configurationService = configurationService;  // 断言工厂初始化，同下面过滤器器工厂类似，也有截去RoutePredicateFactory的操作  initFactories(predicates);  gatewayFilterFactories.forEach(      // factory.name() 用以获取过滤器工厂的名字，具体代码实现就是 末尾截去"GatewayFilterFactory"      // 这也是为什么我们自定义过滤器工厂时需要以GatewayFilterFactory结尾命名      factory -> this.gatewayFilterFactories.put(factory.name(), factory));  this.gatewayProperties = gatewayProperties;}

spring:  cloud:    gateway:      routes:      - id: path_route        uri: https://example.org        predicates:        - Path=/red/{segment},/blue/{segment}        filters:        - AddRequestHeader=X-Request-red, blue

@Bean@Primary // 注意 @Primary 决定了上面被注入的是这个Bean，// 在GatewayAutoConfiguration 中装配了多个RouteDefinitionLocator的子类，包括// CompositeRouteDefinitionLocator、InMemoryRouteDefinitionRepository，// 还在其他配置类中装配的// PropertiesRouteDefinitionLocator 、DiscoveryClientRouteDefinitionLocator public RouteDefinitionLocator routeDefinitionLocator(  List<RouteDefinitionLocator> routeDefinitionLocators) {  return new CompositeRouteDefinitionLocator(Flux.fromIterable(routeDefinitionLocators));}

# 开启此locatorspring.cloud.gateway.discovery.locator.enabled=true# 开启默认为 reactive 模式，需显示关闭可调整为阻塞模式spring.cloud.discovery.reactive.enabled=false

@Document("gwRoutes")public class MongoRouteDefinition extends RouteDefinition {​  public static MongoRouteDefinition from(RouteDefinition route) {    MongoRouteDefinition newRoute = new MongoRouteDefinition();    BeanUtils.copyProperties(route, newRoute);    return newRoute;  }}

public interface RouteRepositoryOperations extends    ReactiveMongoRepository<MongoRouteDefinition, String> {​  /**   * 获取所有路由（分页查询）   *   * @param pageable 分页   * @return 当前页   */  @Query(value = "{}", sort = "{_id:1}")  Flux<MongoRouteDefinition> findAll(Pageable pageable);}

@Slf4j@Componentpublic class MongoRouteDefinitionRepository    implements RouteDefinitionRepository, ApplicationEventPublisherAware {​  private ApplicationEventPublisher eventPublisher;​  private Map<String, RouteDefinition> cache = new ConcurrentHashMap<>();​  private final RouteRepositoryOperations repositoryOperation;​  public MongoRouteDefinitionRepository(RouteRepositoryOperations repositoryOperation) {    this.repositoryOperation = repositoryOperation;  }​  @Override  public Flux<RouteDefinition> getRouteDefinitions() {    return Flux.fromIterable(cache.values());  }​  @Override  public Mono<Void> save(Mono<RouteDefinition> route) {    return route.flatMap(        r -> repositoryOperation.save(MongoRouteDefinition.from(r))            .doOnNext(this::addCache)            .then(Mono.empty())    );  }​  @Override  public Mono<Void> delete(Mono<String> routeId) {    return repositoryOperation.findById(routeId)        .map(RouteDefinition::getId)        .doOnNext(this::removeCache)        .flatMap(repositoryOperation::deleteById);  }​  @Override  public void setApplicationEventPublisher(ApplicationEventPublisher eventPublisher) {    this.eventPublisher = eventPublisher;  }​  /**   * 将指定路由加入到缓存中。   * <p>   * 为了能实时加载路由，可以通过MongoDB的ChangeStream,监听到数据变化后调用此方法   */  public void addCache(RouteDefinition route) {    this.cache.putIfAbsent(route.getId(), route);    // 前文中的CachingRouteLocator会监听此事件，并清空缓存重新从RouteDefinitionLocator中加载路由    this.publishEvent();  }​  /**   * 将指定路由从缓存中删除。   * <p>   * 为了能实时加载路由，可以通过MongoDB的ChangeStream,监听到数据变化后调用此方法   */  public void removeCache(String routeId) {    if (this.cache.remove(routeId) != null) {      this.publishEvent();    }  }​  void publishEvent() {    eventPublisher.publishEvent(new RefreshRoutesEvent(this));  }​  RouteRepositoryOperations getRepositoryOperation() {    return repositoryOperation;  }​​  Map<String, RouteDefinition> getCache() {    return cache;  }​  void setCache(      Map<String, RouteDefinition> cache) {    this.cache = cache;  }}

@Lazy(false)@Component@ConditionalOnProperty(value = "changeStream.enabled", havingValue = "true", matchIfMissing = true)public class RouteChangeStreamHandler {​  private final ReactiveMongoTemplate mongoTemplate;  private final MongoRouteDefinitionRepository routeRepository;​  public RouteChangeStreamHandler(      MongoRouteDefinitionRepository routeRepository, ReactiveMongoTemplate mongoTemplate) {    this.routeRepository = routeRepository;    this.mongoTemplate = mongoTemplate;  }​  @PostConstruct  public void run() {    new Thread(this::startMonitor, "ChangeStream-Monitor-routes").start();  }​  public void startMonitor() {    Aggregation aggregation = Aggregation.newAggregation(Aggregation        .match(Criteria.where("operationType").in("insert", "delete", "update", "replace")));​    ChangeStreamOptions options = ChangeStreamOptions.builder()        .filter(aggregation)        .returnFullDocumentOnUpdate()        .build();​    String collectionName = MongoRouteDefinition.class.getAnnotation(Document.class).value();    Flux<ChangeStreamEvent<MongoRouteDefinition>> changeStream = mongoTemplate        .changeStream(collectionName, options, MongoRouteDefinition.class);​    changeStream        .log()        .doOnNext(e -> {          if (OperationType.INSERT == e.getOperationType()              || OperationType.UPDATE == e.getOperationType()              || OperationType.REPLACE == e.getOperationType()) {            Optional.ofNullable(e.getBody()).ifPresent(routeRepository::addCache);          } else if (OperationType.DELETE == e.getOperationType()) {            getId(e).ifPresent(routeRepository::removeCache);          }        }).blockLast();  }​  private Optional<String> getId(ChangeStreamEvent<MongoRouteDefinition> e) {    return Optional.ofNullable(e.getRaw())        .flatMap(raw -> Optional.ofNullable(raw.getDocumentKey()))        .flatMap(docKey -> Optional.ofNullable(docKey.getObjectId("_id")))        .flatMap(bson -> Optional.of(bson.getValue().toHexString()));  }}

@Slf4j@Component@ConditionalOnProperty(value = "route.schedule.enabled", havingValue = "true", matchIfMissing = true)public class RouteRefresher {​  private final MongoRouteDefinitionRepository repository;​  public RouteRefresher(      MongoRouteDefinitionRepository repository) {    this.repository = repository;  }​  /**   * 固定间隔重新加载一次缓存   */  @Scheduled(initialDelay = 10000, fixedDelay = 60 * 60 * 1001)  private void refresh() {    RouteRepositoryOperations operation = repository.getRepositoryOperation();​    int page = 0;    int pageSize = 1000;    int total = Math.toIntExact(operation.count().blockOptional().orElse(0L));    Map<String, RouteDefinition> oldCache = repository.getCache();    Map<String, RouteDefinition> newCache = new ConcurrentHashMap<>(total);    int oldTotal = oldCache.size();    if (oldTotal < 1) {      // 首次同步刷新      repository.setCache(newCache);    }    while (page * pageSize < total) {      operation.findAll(PageRequest.of(page++, pageSize))          .doOnNext(route -> newCache.putIfAbsent(route.getId(), route))          .blockLast();      log.info("动态路由表当前总大小为：{}, 新路由表当前大小为：{}", oldTotal, newCache.size());    }    repository.setCache(newCache);    log.info("新路由表加载完成，当前大小为：{}", newCache.size());    oldCache.clear();    repository.publishEvent();  }}