这是网关在路由匹配的过程中出现的一小段的代码，刚接触起来还是比较难懂的，:(

核心在于理解

  • Flux和Mono的订阅特性，如果没有订阅，那么内部的这段代码是不运行的. 他的花销仅仅在于实例化整个动作
  • Flux和Mono的每次订阅 subscribe 都会触发它们提前设定的内容

    不恰当的比喻

    Mono: 对比成一个普通的对象，例如 User ，Person 实例对象
    Flux: 对比成一个List，一个流水线，是可以中断的。

每次路由都会进入这段代码进行路由(找到路由)

  1. public class Test1 {
  3.     final static Logger log = LoggerFactory.getLogger(Test1.class);
  4.     private static final String CACHE_KEY = "routes";
  6.     private static Flux<String> fetch() {
  7.         System.out.println("111111");
  8.         return Flux.fromIterable(Arrays.asList("11", "22", "33", "44"));
  9.     }
  11.     private static final Map<String, List> cache = new ConcurrentHashMap<>();
  13.     Flux<String> routes = lookup(cache, CACHE_KEY, String.class).onCacheMissResume(Test1::fetch);
  15.     public static <KEY, VALUE> CacheFlux.FluxCacheBuilderMapMiss<VALUE> lookup(
  16.             Map<KEY, ? super List> cacheMap, KEY key, Class<VALUE> valueClass) {
  17.         return new CacheFlux.FluxCacheBuilderMapMiss<VALUE>() {
  19.             @Override
  20.             public Flux onCacheMissResume(Supplier<Flux<VALUE>> otherSupplier) {
  21.                 return Flux.defer(() -> {
  22.                     Object fromCache = cacheMap.get(key);
  23.                     if (fromCache == null) {
  24.                         return otherSupplier.get()
  25.                                 .materialize()
  26.                                 .collectList()
  27.                                 .doOnNext(signals -> cacheMap.put(key, signals))
  28.                                 .flatMapIterable(Function.identity())
  29.                                 .dematerialize();
  30.                     } else if (fromCache instanceof List) {
  31.                         try {
  32.                             @SuppressWarnings("unchecked")
  33.                             List<Signal<VALUE>> fromCacheSignals = (List<Signal<VALUE>>) fromCache;
  34.                             return Flux.fromIterable(fromCacheSignals)
  35.                                     .dematerialize();
  36.                         } catch (Throwable cause) {
  37.                             return Flux.error(new IllegalArgumentException("Content of cache for key " + key + " cannot be cast to List<Signal>", cause));
  38.                         }
  39.                     } else {
  40.                         return Flux.error(new IllegalArgumentException("Content of cache for key " + key + " is not a List"));
  41.                     }
  42.                 });
  43.             }
  44.         };
  45.     }
  47.     @Test
  48.     public void main() {
  49.         Flux.fromIterable(Arrays.asList("55", "66", "77", "88"))
  50.                 .materialize()
  51.                 .collectList()
  52.                 .doOnNext(x -> cache.put(CACHE_KEY, x)).subscribe();
  53.         for (int i = 0; i < 10; i++) {
  54.             int finalI = i;
  55.             //每次都会进入onCacheMissResume
  56.             final Mono<String> map = routes
  57.                     .concatMap(x -> Mono.just(x).filterWhen(r -> {
  58.                         return Mono.just(r.equals(finalI + "" + finalI));
  59.                     })
  60.                             .doOnError(e -> log.error("ff"))
  61.                             .onErrorResume(e -> Mono.empty()))
  62.                     .next()
  63.                     .map(route -> {
  64.                         return route;
  65.                     });
  66.             map.subscribe(System.out::println);
  67.         }
  69.     }
  70. }

引申

  • 如何实现动态路由呢?
    • 外接注册中心，例如Nacos
      • 2020年的时候我对SpringCloud整体还不够熟悉，实际上springCloud Gateway已经实现了，详见 ReactiveCompositeDiscoveryClientAutoConfiguration.java
  • 如何快速的实现路由过滤呢
    • 网关本身只是作为 源服务 的一个代理，那么本身的消耗应该是微乎其微的.
    • SpringCloud Gateway是通过List遍历的形式实现的，利用是Java8的 Predicate 函数。

      参考链接