Spring Cloud Gateway 源码浅析

 1@SpringBootApplication
 2public class DatanexusGatewayApplication {
 3    public static void main(String[] args) {
 4        SpringApplication.run(GatewayApplication.class, args);
 5    }
 6}
 7
 8//1、构建 SpringApplication 实例
 9public SpringApplication(ResourceLoader resourceLoader, Class<?>... primarySources) {
10    this.resourceLoader = resourceLoader;
11    Assert.notNull(primarySources, "PrimarySources must not be null");
12    this.primarySources = new LinkedHashSet<>(Arrays.asList(primarySources));
13    //1.1、判断classpath是否存在org.springframework.web.reactive.DispatcherHandler类，存在说明是 		  WebApplicationType.REACTIVE应用类型
14    this.webApplicationType = WebApplicationType.deduceFromClasspath();
15    this.bootstrapRegistryInitializers = new ArrayList<>(
16          getSpringFactoriesInstances(BootstrapRegistryInitializer.class));
17    //1.2、从META-INF中获取应用上下文初始化
18    setInitializers((Collection) getSpringFactoriesInstances(ApplicationContextInitializer.class));
19    //1.3、从META-INF中获应用监听器
20    setListeners((Collection) getSpringFactoriesInstances(ApplicationListener.class));
21    //1.4、根据当前stackTrace信息反射获取当前运行的ApplicationClass
22    this.mainApplicationClass = deduceMainApplicationClass();
23}
24
25//2、调用 SpringApplication run()方法，启动 SpringApplication
26public ConfigurableApplicationContext run(String... args) {
27		long startTime = System.nanoTime();
28  	//2.1、创建默认的启动上下文
29		DefaultBootstrapContext bootstrapContext = createBootstrapContext();
30		ConfigurableApplicationContext context = null;
31		configureHeadlessProperty();
32  
33  	//2.2、通过spring boot自动装配获取Spring应用运行监听器并启动（扫描META-INF/factories注册的类）
34		SpringApplicationRunListeners listeners = getRunListeners(args);
35		listeners.starting(bootstrapContext, this.mainApplicationClass);
36  	
37		try {
38      //2.3、封装命令行输入的应用参数
39			ApplicationArguments applicationArguments = new DefaultApplicationArguments(args);
40      
41      //2.4、根据webApplication类型构建可配置的环境变量实例
42			ConfigurableEnvironment environment = prepareEnvironment(listeners, bootstrapContext, applicationArguments);
43      //2.5、配置可忽略的Bean信息
44			configureIgnoreBeanInfo(environment);
45      
46      //2.6、构建BannerPrinter打印自定义的信息
47			Banner printedBanner = printBanner(environment);
48      
49      //2.7、根据webApplicationType构建应用上下文，gateway使用的是AnnotationConfigReactiveWebApplicationContext
50			context = createApplicationContext();
51			context.setApplicationStartup(this.applicationStartup);
52      
53      //2.8、完善上下文，把启动上下文、环境实例、应用监听器、BannerPrinter与上下文关联
54			prepareContext(bootstrapContext, context, environment, listeners, applicationArguments, printedBanner);
55      
56      //2.9、重点、重点、重点：刷新上下文，触发初始化Spring Cloud Gateway等组件资源的初始化，下面重点分析
57      //通过模板设计模式，该接口会调用抽象类AbstractApplicationContext的refresh()方法，再根据webApplicationType类型调用不同的子类，此处是ReactiveWebServerApplicationContext.refresh()方法
58			refreshContext(context);
59      
60      //3.0、刷新后操作，目前暂无实现
61			afterRefresh(context, applicationArguments);
62			Duration timeTakenToStartup = Duration.ofNanos(System.nanoTime() - startTime);
63			if (this.logStartupInfo) {
64				new StartupInfoLogger(this.mainApplicationClass).logStarted(getApplicationLog(), timeTakenToStartup);
65			}
66      //3.1、启动应用监听器
67			listeners.started(context, timeTakenToStartup);
68      
69      //3.2、执行runner的run逻辑
70			callRunners(context, applicationArguments);
71		}
72		catch (Throwable ex) {
73			//省略其余代码......
74		}
75			//省略其余代码......
76		return context;
77	}

 1//代码位置：GatewayAutoConfiguration
 2@Bean
 3public RouteLocatorBuilder routeLocatorBuilder(ConfigurableApplicationContext context) {
 4  return new RouteLocatorBuilder(context);
 5}
 6
 7//默认创建 Properties 配置的路由定义定位器实例
 8@Bean
 9@ConditionalOnMissingBean
10public PropertiesRouteDefinitionLocator propertiesRouteDefinitionLocator(GatewayProperties properties) {
11  return new PropertiesRouteDefinitionLocator(properties);
12}
13
14//默认创建基于内存保存的路由定义仓库实例
15@Bean
16@ConditionalOnMissingBean(RouteDefinitionRepository.class)
17public InMemoryRouteDefinitionRepository inMemoryRouteDefinitionRepository() {
18  return new InMemoryRouteDefinitionRepository();
19}
20
21//创建具有组合能力的路由定义定位器，此处routeDefinitionLocators 列表元素为InMemoryRouteDefinitionRepository、PropertiesRouteDefinitionLocator
22@Bean
23@Primary
24public RouteDefinitionLocator routeDefinitionLocator(List<RouteDefinitionLocator> routeDefinitionLocators) {
25  return new CompositeRouteDefinitionLocator(Flux.fromIterable(routeDefinitionLocators));
26}
27
28//创建路由定位器实例
29@Bean
30public RouteLocator routeDefinitionRouteLocator(GatewayProperties properties,
31      List<GatewayFilterFactory> gatewayFilters, List<RoutePredicateFactory> predicates,
32      RouteDefinitionLocator routeDefinitionLocator, ConfigurationService configurationService) {
33   //初始化用户配置的路由、断言等配置信息
34   return new RouteDefinitionRouteLocator(routeDefinitionLocator, predicates, gatewayFilters, properties,
35         configurationService);
36}
37
38//创建具有缓存组合能力的路由定位器
39@Bean
40@Primary
41@ConditionalOnMissingBean(name = "cachedCompositeRouteLocator")
42public RouteLocator cachedCompositeRouteLocator(List<RouteLocator> routeLocators) {
43   //
44   return new CachingRouteLocator(new CompositeRouteLocator(Flux.fromIterable(routeLocators)));
45}

 1public class RouteRefreshListener implements ApplicationListener<ApplicationEvent> {
 2
 3  private final ApplicationEventPublisher publisher;
 4
 5  public RouteRefreshListener(ApplicationEventPublisher publisher) {
 6    Assert.notNull(publisher, "publisher may not be null");
 7    this.publisher = publisher;
 8  }
 9
10  @Override
11  public void onApplicationEvent(ApplicationEvent event) {
12    //当前是上下文刷新事件且当前刷新的应用上下文没有 management 命名空间，则发布 RefreshRoutesEvent 事件
13    if (event instanceof ContextRefreshedEvent) {
14      ContextRefreshedEvent refreshedEvent = (ContextRefreshedEvent) event;
15      if (!WebServerApplicationContext.hasServerNamespace(refreshedEvent.getApplicationContext(), "management")) {
16        reset();
17      }
18    }
19    else if (event instanceof RefreshScopeRefreshedEvent || event instanceof InstanceRegisteredEvent) {
20      reset();
21    }
22    //省略部分代码......
23  }
24	
25  //根据参数判断是否需发布 RefreshRoutesEvent 事件
26  private void resetIfNeeded(Object value) {
27    if (this.monitor.update(value)) {
28      reset();
29    }
30  }
31
32  //发布 RefreshRoutesEvent 事件，CachingRouteLocator 监听器会监听到事件
33  private void reset() {
34    this.publisher.publishEvent(new RefreshRoutesEvent(this));
35  }
36}

 1public class CachingRouteLocator
 2      implements Ordered, RouteLocator, ApplicationListener<RefreshRoutesEvent>, ApplicationEventPublisherAware {
 3  //此处省略部分代码.....
 4
 5  //拉取路由配置信息，并按照 order 优先级排序,此处的 delegate 是 CompositeRouteLocator 实例
 6   private Flux<Route> fetch() {
 7      return this.delegate.getRoutes().sort(AnnotationAwareOrderComparator.INSTANCE);
 8   }
 9
10  //监听 RefreshRoutesEvent 事件
11   @Override 
12   public void onApplicationEvent(RefreshRoutesEvent event) {
13      try {
14         //拉取路由配置信息
15         fetch().collect(Collectors.toList()).subscribe(
16               list -> Flux.fromIterable(list).materialize().collect(Collectors.toList()).subscribe(signals -> {
17                  //发布 RefreshRoutesResultEvent 事件
18                  applicationEventPublisher.publishEvent(new RefreshRoutesResultEvent(this));
19                  cache.put(CACHE_KEY, signals);
20               }, this::handleRefreshError), this::handleRefreshError);
21      }
22      catch (Throwable e) {
23         handleRefreshError(e);
24      }
25   }
26
27   private void handleRefreshError(Throwable throwable) {
28      //省略部分代码.....
29      applicationEventPublisher.publishEvent(new RefreshRoutesResultEvent(this, throwable));
30   }
31  
32   //省略部分代码.....
33}

 1//代码位置：RouteDefinitionRouteLocator.getRoutes()
 2@Override
 3public Flux<Route> getRoutes() {
 4  //此处的 routeDefinitionLocator 为 CompositeRouteDefinitionLocator 实例
 5   Flux<Route> routes = 	this.routeDefinitionLocator.getRouteDefinitions().map(this::convertToRoute);
 6   //省略部分代码....  
 7   return routes.map(route -> {
 8      if (logger.isDebugEnabled()) {
 9         logger.debug("RouteDefinition matched: " + route.getId());
10      }
11      return route;
12   });
13}
14
15private Route convertToRoute(RouteDefinition routeDefinition) {
16  AsyncPredicate<ServerWebExchange> predicate = combinePredicates(routeDefinition);
17  List<GatewayFilter> gatewayFilters = getFilters(routeDefinition);
18
19  return Route.async(routeDefinition).asyncPredicate(predicate).replaceFilters(gatewayFilters).build();
20}

 1//获取 spring.cloud.gateway 配置的 Route Definitions
 2public class PropertiesRouteDefinitionLocator implements RouteDefinitionLocator {
 3
 4 private final GatewayProperties properties;
 5
 6 @Override
 7 public Flux<RouteDefinition> getRouteDefinitions() {
 8    return Flux.fromIterable(this.properties.getRoutes());
 9 }
10}
11
12// Route Definition 内存仓库
13public class InMemoryRouteDefinitionRepository implements RouteDefinitionRepository {
14	private final Map<String, RouteDefinition> routes = synchronizedMap(new LinkedHashMap<String, RouteDefinition>());
15   	
16  //省略部分代码.....
17	@Override
18	public Flux<RouteDefinition> getRouteDefinitions() {
19		Map<String, RouteDefinition> routesSafeCopy = new LinkedHashMap<>(routes);
20		return Flux.fromIterable(routesSafeCopy.values());
21	}
22}

 1//AbstractRefreshableApplicationContext.onRefresh()
 2@Override
 3public void refresh() throws BeansException, IllegalStateException {
 4  synchronized (this.startupShutdownMonitor) {
 5    StartupStep contextRefresh = this.applicationStartup.start("spring.context.refresh");
 6
 7    // 1、刷新上下文预准备操作，包括设置上下文状态、初始化占位符属性源、校验必要属性
 8    prepareRefresh();
 9
10    // 2、通知子类刷新 bean 工厂类
11    ConfigurableListableBeanFactory beanFactory = obtainFreshBeanFactory();
12
13    // 3、准备用于上下文使用的 bean 工厂
14    prepareBeanFactory(beanFactory);
15
16    try {
17      //4、允许在上下文子类中对 bean 工程进行后处理操作
18      postProcessBeanFactory(beanFactory);
19
20      StartupStep beanPostProcess = this.applicationStartup.start("spring.context.beans.post-process");
21      //5、调用并初始化在上下文中注册为 bean 的工厂处理器
22      invokeBeanFactoryPostProcessors(beanFactory);
23
24      // 6、注册拦截 bean 创建的 bean 处理器
25      registerBeanPostProcessors(beanFactory);
26      beanPostProcess.end();
27
28      //7、为上下文初始化 message 源
29      initMessageSource();
30
31      //8、为上下文初始化事件组播期器
32      initApplicationEventMulticaster();
33
34      //9、在特定上下文子类中初始化其他特定的 bean,spring cloud gateway 的路由、过滤器等在该子类中实现
35      onRefresh();
36
37      //10、检查并注册监听器 bean
38      registerListeners();
39
40      //11、实例化所有剩余的（非延迟初始化）的单例
41      finishBeanFactoryInitialization(beanFactory);
42
43      //12、初始化生命周期处理器、发布 ContextRefreshedEvent 事件
44      finishRefresh();
45    }
46    catch (BeansException ex) {
47      //省略部分代码......
48      throw ex;
49    }
50    finally {
51      //省略部分代码......
52    }
53  }
54}

 1//ReactiveWebServerApplicationContext.onRefresh()
 2@Override
 3protected void onRefresh() {
 4  super.onRefresh();
 5  try {
 6    createWebServer();
 7  }
 8  //省略部分代码......
 9}
10
11private void createWebServer() {
12		WebServerManager serverManager = this.serverManager;
13  	//第一次创建时，serverManager 是为空的
14		if (serverManager == null) {
15      //1、记录webserver创建的步骤
16			StartupStep createWebServer = this.getApplicationStartup().start("spring.boot.webserver.create");
17      //2、获取 WebServerFactory bean 名称，gateway 对应的 bean 名称 nettyReactiveWebServerFactory
18			String webServerFactoryBeanName = getWebServerFactoryBeanName();
19			ReactiveWebServerFactory webServerFactory = getWebServerFactory(webServerFactoryBeanName);
20			createWebServer.tag("factory", webServerFactory.getClass().toString());
21     
22			boolean lazyInit = getBeanFactory().getBeanDefinition(webServerFactoryBeanName).isLazyInit();
23      //3、创建 webServerManager 实例，getHttpHandler()方法获取绑定处理 http 请求的 hanlder
24			this.serverManager = new WebServerManager(this, webServerFactory, this::getHttpHandler, lazyInit);
25      //4、注册 webServerGracefulShutdown bean,优美地关闭 webserver
26			getBeanFactory().registerSingleton("webServerGracefulShutdown",
27					new WebServerGracefulShutdownLifecycle(this.serverManager.getWebServer()));
28      
29      //5、注册 webServerStartStop bean,优美地启动或停止 webserver
30			getBeanFactory().registerSingleton("webServerStartStop",
31					new WebServerStartStopLifecycle(this.serverManager));
32			createWebServer.end();
33		}
34  	//6、用实际的实例，替换任何属性源
35		initPropertySources();
36	}

 1//ReactiveWebServerApplicationContext.getHttpHandler(),返回 httpHandler 请求处理器
 2protected HttpHandler getHttpHandler() {
 3  // 这里使用 bean 名称是可以避免层级结构
 4  String[] beanNames = getBeanFactory().getBeanNamesForType(HttpHandler.class);
 5  if (beanNames.length == 0) {
 6    throw new ApplicationContextException(
 7        "Unable to start ReactiveWebApplicationContext due to missing HttpHandler bean.");
 8  }
 9  if (beanNames.length > 1) {
10    throw new ApplicationContextException(
11        "Unable to start ReactiveWebApplicationContext due to multiple HttpHandler beans : "
12            + StringUtils.arrayToCommaDelimitedString(beanNames));
13  }
14  //委派模式，HttpWebHandlerAdapter -> ExceptionHandlingWebHandler -> FilteringWebHandler -> DispatcherHandler
15  return getBeanFactory().getBean(beanNames[0], HttpHandler.class);
16}

 1//	WebServerManager 构造器，在 factory.getWebServer(this.handler) 进一步创建 nettyWebServer
 2// factory 变量实际引用的是 NettyReactiveWebServerFactory
 3WebServerManager(ReactiveWebServerApplicationContext applicationContext, ReactiveWebServerFactory factory,
 4    Supplier<HttpHandler> handlerSupplier, boolean lazyInit) {
 5		this.applicationContext = applicationContext;
 6		Assert.notNull(factory, "Factory must not be null");
 7		this.handler = new DelayedInitializationHttpHandler(handlerSupplier, lazyInit);
 8		this.webServer = factory.getWebServer(this.handler);
 9}
10
11// NettyReactiveWebServerFactory.getWebServer(HttpHandler httpHandler)，该处的 httpHandler 具体是实现为 DelayedInitializationHttpHandler，该 httpHandler 实例化发生在 WebServerManager.start()方法。
12@Override
13public WebServer getWebServer(HttpHandler httpHandler) {
14  	//1、创建 httpServer
15		HttpServer httpServer = createHttpServer();
16  	//2、创建 httpHandler 适配器对象
17		ReactorHttpHandlerAdapter handlerAdapter = new ReactorHttpHandlerAdapter(httpHandler);
18  	//3、创建 nettyWebServer，把 httpServer、httpHandler 统一封装到这个实例中
19		NettyWebServer webServer = createNettyWebServer(httpServer, handlerAdapter, this.lifecycleTimeout,
20				getShutdown());
21		webServer.setRouteProviders(this.routeProviders);
22		return webServer;
23}
24
25// NettyReactiveWebServerFactory.createHttpServer()
26private HttpServer createHttpServer() {
27		HttpServer server = HttpServer.create();
28		if (this.resourceFactory != null) {
29			LoopResources resources = this.resourceFactory.getLoopResources();
30			Assert.notNull(resources, "No LoopResources: is ReactorResourceFactory not initialized yet?");
31      //绑定监听地址，获取 ServerProperties 配置的端口，详细可参考NettyWebServerFactoryCustomizer，此处的 getListenAddress 参数是 Supplier 类型，在使用时才会真正调用方法
32			server = server.runOn(resources).bindAddress(this::getListenAddress);
33		}
34		else {
35			server = server.bindAddress(this::getListenAddress);
36		}
37		if (getSsl() != null && getSsl().isEnabled()) {
38			server = customizeSslConfiguration(server);
39		}
40		if (getCompression() != null && getCompression().getEnabled()) {
41			CompressionCustomizer compressionCustomizer = new CompressionCustomizer(getCompression());
42			server = compressionCustomizer.apply(server);
43		}
44		server = server.protocol(listProtocols()).forwarded(this.useForwardHeaders);
45		return applyCustomizers(server);
46}

 1// 代码位置：AbstractApplicationContext.finishRefresh()
 2protected void finishRefresh() {
 3  //1、清理上下文层级的资源缓存，例如来自扫描的 ASM 原数据
 4  clearResourceCaches();
 5
 6  //2、为该上下文初始化生命周期处理器，该处使用的生命周期处理器是 DefaultLifecycleProcessor
 7  initLifecycleProcessor();
 8
 9  //3、将刷新传播到生命周期处理器，重点分析：调用生命周期处理器的 start()方法，
10  getLifecycleProcessor().onRefresh();
11
12  //4、发布上下文刷新事件 ContextRefreshedEvent
13  publishEvent(new ContextRefreshedEvent(this));
14
15  //5、探测 GraalVM native image 环境是否开启，若开启则注册到应用上下文中
16  if (!NativeDetector.inNativeImage()) {
17    LiveBeansView.registerApplicationContext(this);
18  }
19}

 1// 代码位置：DefaultLifecycleProcessor.onRefresh()
 2@Override
 3public void onRefresh() {
 4  startBeans(true);
 5  this.running = true;
 6}
 7
 8private void startBeans(boolean autoStartupOnly) {
 9  //1、从 beanFactory 工厂中获取当前生命周期实例,格式<beanName,生命周期实例>
10  Map<String, Lifecycle> lifecycleBeans = getLifecycleBeans();
11  Map<Integer, LifecycleGroup> phases = new TreeMap<>();
12	
13  //2、遍历并计算生命周期实例阶段值 phase
14  lifecycleBeans.forEach((beanName, bean) -> {
15    if (!autoStartupOnly || (bean instanceof SmartLifecycle && ((SmartLifecycle) bean).isAutoStartup())) {
16      int phase = getPhase(bean);
17      phases.computeIfAbsent(
18          phase,
19          p -> new LifecycleGroup(phase, this.timeoutPerShutdownPhase, lifecycleBeans, autoStartupOnly)
20      ).add(beanName, bean);
21    }
22  });
23  //3、循环调用生命周期组的 start 方法
24  if (!phases.isEmpty()) {
25    phases.values().forEach(LifecycleGroup::start);
26  }
27}
28
29public void start() {
30  //省略部分代码
31  for (LifecycleGroupMember member : this.members) {
32    doStart(this.lifecycleBeans, member.name, this.autoStartupOnly);
33  }
34}
35
36private void doStart(Map<String, ? extends Lifecycle> lifecycleBeans, String beanName, boolean autoStartupOnly) {
37  Lifecycle bean = lifecycleBeans.remove(beanName);
38  if (bean != null && bean != this) {
39    //1、如果该 lifecycleBean 存在依赖实例，则递归完成该依赖实例的 start()方法
40    String[] dependenciesForBean = getBeanFactory().getDependenciesForBean(beanName);
41    for (String dependency : dependenciesForBean) {
42      doStart(lifecycleBeans, dependency, autoStartupOnly);
43    }
44    
45    //2、检测生命周期 bean 实例是否处于 running 状态
46    if (!bean.isRunning() &&
47        (!autoStartupOnly || !(bean instanceof SmartLifecycle) || ((SmartLifecycle) bean).isAutoStartup())) {
48      if (logger.isTraceEnabled()) {
49        logger.trace("Starting bean '" + beanName + "' of type [" + bean.getClass().getName() + "]");
50      }
51      try {
52        //通过模版设计模式，调用不同的子类实现 start() 方法
53        bean.start();
54      }
55      catch (Throwable ex) {
56        throw new ApplicationContextException("Failed to start bean '" + beanName + "'", ex);
57      }
58      if (logger.isDebugEnabled()) {
59        logger.debug("Successfully started bean '" + beanName + "'");
60      }
61    }
62  }
63}

 1// 代码位置：WebServerStartStopLifecycle.start()
 2@Override
 3public void start() {
 4  //调用组合的 WebServerManager 实例启动方法
 5  this.weServerManager.start();
 6  this.running = true;
 7}
 8
 9// 代码位置：WebServerManager.start()
10void start() {
11  //初始化 httpHandler
12  this.handler.initializeHandler();
13  
14  //启动 webServer,最终调用 NettyWebServer.startHttpServer()
15  this.webServer.start();
16  
17  //发布 ReactiveWebServerInitializedEvent 事件
18  this.applicationContext
19      .publishEvent(new ReactiveWebServerInitializedEvent(this.webServer, this.applicationContext));
20}

 1//代码位置：NettyWebServer.startHttpServer()
 2DisposableServer startHttpServer() {
 3   HttpServer server = this.httpServer;
 4   if (this.routeProviders.isEmpty()) {
 5     	//绑定 server 对应的 workHandler
 6      server = server.handle(this.handler);
 7   }
 8   else {
 9      server = server.route(this::applyRouteProviders);
10   }
11   if (this.lifecycleTimeout != null) {
12      return server.bindNow(this.lifecycleTimeout);
13   }
14   //将 channel、handler 等绑定到 nettyServer 上,详细实现代码可查看：ServerTransport.bind
15   return server.bindNow();
16}
17
18//代码位置：HttpServer.handle()
19	public final HttpServer handle(
20			BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> handler) {
21		Objects.requireNonNull(handler, "handler");
22		return childObserve(new HttpServerHandle(handler));
23	}

 1@Override
 2public void channelRead(ChannelHandlerContext ctx, Object msg) {
 3   //省略部分的代码......
 4   // read message and track if it was keepAlive
 5   if (msg instanceof HttpRequest) {
 6      IdleTimeoutHandler.removeIdleTimeoutHandler(ctx.pipeline());
 7
 8      final HttpRequest request = (HttpRequest) msg;
 9    	//省略部分的代码......
10
11       HttpServerOperations ops;
12       try {
13          //构建 HttpServerOperations 对象，后面作为request、response 使用
14          ops = new HttpServerOperations(Connection.from(ctx.channel()),
15                listener,request,compress,
16                ConnectionInfo.from(ctx.channel(),request,secure,remoteAddress,
17         forwardedHeaderHandler),cookieDecoder,cookieEncoder,formDecoderProvider,mapHandle,secure);
18         }
19         catch (RuntimeException e) {
20            sendDecodingFailures(e, msg);
21            return;
22         }
23         ops.bind();
24         //第一次请求时会先传递 ConnectionObserver.State.CONFIGURED 事件给 ServerTransport 类处理，随后传递至 ReactorNetty.CompositeConnectionObserver 类，最后会执行下述的 ctx.fireChannelRead
25         listener.onStateChange(ops, ConnectionObserver.State.CONFIGURED);
26				 //从当前节点往下传播事件
27         ctx.fireChannelRead(msg);
28         return;
29   }
30	//省略部分的代码......
31   ctx.fireChannelRead(msg);
32}

 1//代码位置：ChannelOperationsHandler.channelRead(ChannelHandlerContext ctx, Object msg)
 2@Override
 3final public void channelRead(ChannelHandlerContext ctx, Object msg) {
 4	//省略部分代码.....
 5  ChannelOperations<?, ?> ops = ChannelOperations.get(ctx.channel());
 6  if (ops != null) {
 7    //调用 HttpServerOperations 类的 onInboundNext 方法
 8    ops.onInboundNext(ctx, msg);
 9  }
10    //省略部分代码.....
11}
12
13//代码位置：HttpServerOperations.onInboundNext(ChannelHandlerContext ctx, Object msg)
14@Override
15protected void onInboundNext(ChannelHandlerContext ctx, Object msg) {
16   if (msg instanceof HttpRequest) {
17      //生成 HttpServerState.REQUEST_RECEIVED 事件，调用 ConnectionObserver 的 onStateChange 
18       listener().onStateChange(this, HttpServerState.REQUEST_RECEIVED);
19      //省略部分代码.....
20      return;
21   }
22			//省略部分代码.....
23}

 1public void onStateChange(Connection connection, State newState) {
 2   if (newState == HttpServerState.REQUEST_RECEIVED) {
 3      try {
 4				 //忽略部分代码.....
 5         HttpServerOperations ops = (HttpServerOperations) connection;
 6         //在此处把请求传递到 ReactorHttpHandlerAdapter 适配器
 7         Publisher<Void> publisher = handler.apply(ops, ops);
 8         Mono<Void> mono = Mono.deferContextual(ctx -> {
 9            ops.currentContext = Context.of(ctx);
10            return Mono.fromDirect(publisher);
11         });
12         if (ops.mapHandle != null) {
13            mono = ops.mapHandle.apply(mono, connection);
14         }
15         mono.subscribe(ops.disposeSubscriber());
16      }
17      //忽略部分代码.....
18   }
19}

 1//代码位置：WebHttpHandlerBuilder.build
 2public HttpHandler build() {
 3		WebHandler decorated = new FilteringWebHandler(this.webHandler, this.filters);
 4		decorated = new ExceptionHandlingWebHandler(decorated,  this.exceptionHandlers);
 5		HttpWebHandlerAdapter adapted = new HttpWebHandlerAdapter(decorated);
 6  	//省略部分代码.....
 7		return (this.httpHandlerDecorator != null ? this.httpHandlerDecorator.apply(adapted) : adapted);
 8}
 9	
10	//代码位置：HttpHandlerAutoConfiguration.httpHandler
11	@Bean
12	public HttpHandler httpHandler(ObjectProvider<WebFluxProperties> propsProvider) {
13		HttpHandler httpHandler = 	WebHttpHandlerBuilder.applicationContext(this.applicationContext).build();
14  //省略部分代码.....
15		return httpHandler;
16	}

1@Override
2public Mono<Void> filter(ServerWebExchange exchange) {
3   //判断当前责任链里是否还有过滤器，存在过滤器则执行 filter 方法，没有则执行 DispatcherHandler 的 handle
4   return Mono.defer(() ->
5         this.currentFilter != null && this.chain != null ?
6               invokeFilter(this.currentFilter, this.chain, exchange) :
7               this.handler.handle(exchange));
8}

 1//DispatcherHandler
 2@Override
 3public void setApplicationContext(ApplicationContext applicationContext) {
 4  initStrategies(applicationContext);
 5}
 6
 7protected void initStrategies(ApplicationContext context) {
 8   //1、获取指定 HandlerMapping 类型的所有处理器映射 Bean
 9   Map<String, HandlerMapping> mappingBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
10         context, HandlerMapping.class, true, false);
11	 
12   ArrayList<HandlerMapping> mappings = new ArrayList<>(mappingBeans.values());
13   //2、根据 HandlerMapping 处理器映射（实现 Ordered接口） 优先级排序
14   AnnotationAwareOrderComparator.sort(mappings);
15   this.handlerMappings = Collections.unmodifiableList(mappings);
16	
17   //3、获取指定 HandlerAdapter 类型的所有处理器适配器 Bean
18   Map<String, HandlerAdapter> adapterBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
19         context, HandlerAdapter.class, true, false);
20	
21   this.handlerAdapters = new ArrayList<>(adapterBeans.values());
22   //4、根据 HandlerAdapter 处理器适配器（实现 Ordered接口） 优先级排序
23   AnnotationAwareOrderComparator.sort(this.handlerAdapters);
24	
25   //5、获取指定 HandlerResultHandler 类型的所有结果处理器 Bean
26   Map<String, HandlerResultHandler> beans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
27         context, HandlerResultHandler.class, true, false);
28
29   this.resultHandlers = new ArrayList<>(beans.values());
30   AnnotationAwareOrderComparator.sort(this.resultHandlers);
31}

 1@Override
 2public Mono<Void> handle(ServerWebExchange exchange) {
 3   //1、如果处理器映射为空直接返回错误
 4   if (this.handlerMappings == null) {
 5      return createNotFoundError();
 6   }
 7   //2、判断是否 cors 预请求并处理
 8   if (CorsUtils.isPreFlightRequest(exchange.getRequest())) {
 9      return handlePreFlight(exchange);
10   }
11   //3、遍历处理器映射，并获取处理器映射关联的 webHandler 处理请求、返回结果
12   return Flux.fromIterable(this.handlerMappings)
13         .concatMap(mapping -> mapping.getHandler(exchange))
14         //取出找到的第一个 handler，实际上由 RoutePredicateHandlerMapping 返回 FilteringWebHandler
15         .next()
16     		 //如果为空则返回 not found error
17         .switchIfEmpty(createNotFoundError())
18          //调用 handler 
19         .flatMap(handler -> invokeHandler(exchange, handler))
20         .flatMap(result -> handleResult(exchange, result));
21}

 1//代码位置 AbstractHandlerMapping.getHandler(ServerWebExchange exchange) ,根据
 2@Override
 3public Mono<Object> getHandler(ServerWebExchange exchange) {
 4   // getHandlerInternal(exchange) 是由各子类实现的，我们重点分析网关路由的实现
 5   return getHandlerInternal(exchange).map(handler -> {
 6     	//检查当前的 handler 是否为 CorsConfiguration 或者 当前请求是 cors 预请求
 7      ServerHttpRequest request = exchange.getRequest();
 8      if (hasCorsConfigurationSource(handler) || CorsUtils.isPreFlightRequest(request)) {
 9         CorsConfiguration config = (this.corsConfigurationSource != null ?
10               this.corsConfigurationSource.getCorsConfiguration(exchange) : null);
11         CorsConfiguration handlerConfig = getCorsConfiguration(handler, exchange);
12         config = (config != null ? config.combine(handlerConfig) : handlerConfig);
13         if (config != null) {
14            config.validateAllowCredentials();
15         }
16         // corsProcessor 处理器处理当前请求
17         if (!this.corsProcessor.process(config, exchange) || CorsUtils.isPreFlightRequest(request)) {
18            return NO_OP_HANDLER;
19         }
20      }
21      return handler;
22   });
23}

 1//代码位置：RoutePredicateHandlerMapping.getHandlerInternal(ServerWebExchange exchange)
 2@Override
 3protected Mono<?> getHandlerInternal(ServerWebExchange exchange) {
 4 //1、如果设置的 management 端口与服务器端口不同，则不处理在 manager 端口上的请求
 5 if (this.managementPortType == DIFFERENT && this.managementPort != null
 6       && exchange.getRequest().getURI().getPort() == this.managementPort) {
 7    return Mono.empty();
 8 }
 9 //2、根据请求查找对应的 Route 路由信息，如果没有找到路由信息则返回空，即 Mono.empty()
10 return lookupRoute(exchange)
11       .flatMap((Function<Route, Mono<?>>) r -> {
12          exchange.getAttributes().remove(GATEWAY_PREDICATE_ROUTE_ATTR);
13          //省略部分代码......
14          return Mono.just(webHandler);
15       }).switchIfEmpty(Mono.empty().then(Mono.fromRunnable(() -> {
16          //省略部分代码......
17       })));
18}
19
20//根据请求查找对应的路由
21protected Mono<Route> lookupRoute(ServerWebExchange exchange) {
22  return this.routeLocator.getRoutes()
23      .concatMap(route -> Mono.just(route).filterWhen(r -> {
24        exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR, r.getId());
25        //找到路由后获取它的 predicates 配置信息，并判断当前请求是否匹配
26        return r.getPredicate().apply(exchange);
27      })
28          .doOnError(e -> logger.error("Error applying predicate for route: " + route.getId(), e))
29          .onErrorResume(e -> Mono.empty()))
30    	//省略部分代码......
31      .next()
32      .map(route -> {
33    	//省略部分代码......
34        validateRoute(route, exchange);
35        return route;
36      });
37}

 1spring:
 2  cloud:
 3    gateway:
 4      routes:
 5        - id: route-test
 6          uri: http://127.0.0.1:8080
 7          predicates:
 8            - Path=/user/**
 9          filters:
10            - Authentication

 1//代码位置：PathRoutePredicateFactory.apply(Config config)
 2@Override
 3public Predicate<ServerWebExchange> apply(Config config) {
 4  //解析 yaml 配置中 path predicate 的值，并保存到 pathPatterns 变量
 5  final ArrayList<PathPattern> pathPatterns = new ArrayList<>();
 6  synchronized (this.pathPatternParser) {
 7    pathPatternParser.setMatchOptionalTrailingSeparator(config.isMatchTrailingSlash());
 8    config.getPatterns().forEach(pattern -> {
 9      PathPattern pathPattern = this.pathPatternParser.parse(pattern);
10      pathPatterns.add(pathPattern);
11    });
12  }
13 return new GatewayPredicate() {
14    @Override
15    public boolean test(ServerWebExchange exchange) {
16       //获取当前请求的路径 path
17       PathContainer path = parsePath(exchange.getRequest().getURI().getRawPath());
18       PathPattern match = null;
19       //判断当前请求的路径是否与配置的匹配
20       for (int i = 0; i < pathPatterns.size(); i++) {
21          PathPattern pathPattern = pathPatterns.get(i);
22          if (pathPattern.matches(path)) {
23             match = pathPattern;
24             break;
25          }
26       }
27
28       if (match != null) {
29          traceMatch("Pattern", match.getPatternString(), path, true);
30         	//如果存在环境变量，则解析变量值并放置到请求的 attribute 属性
31          PathMatchInfo pathMatchInfo = match.matchAndExtract(path);
32          putUriTemplateVariables(exchange, pathMatchInfo.getUriVariables());
33          exchange.getAttributes().put(GATEWAY_PREDICATE_MATCHED_PATH_ATTR, match.getPatternString());
34          String routeId = (String) exchange.getAttributes().get(GATEWAY_PREDICATE_ROUTE_ATTR);
35          if (routeId != null) {
36             // populated in RoutePredicateHandlerMapping
37             exchange.getAttributes().put(GATEWAY_PREDICATE_MATCHED_PATH_ROUTE_ID_ATTR, routeId);
38          }
39          return true;
40       }
41       //省略部分代码.....
42    }
43   		//省略部分代码.....
44 };
45}

 1//代码位置：FilteringWebHandler.handle(ServerWebExchange exchange)
 2@Override
 3public Mono<Void> handle(ServerWebExchange exchange) {
 4   //从 ServerWebExchange 上下文查询当前的 Route 路由实例
 5   Route route = exchange.getRequiredAttribute(GATEWAY_ROUTE_ATTR);
 6   //获取路由配置的过滤器
 7   List<GatewayFilter> gatewayFilters = route.getFilters();
 8		
 9   List<GatewayFilter> combined = new ArrayList<>(this.globalFilters);
10   //组合系统默认的过滤器
11   combined.addAll(gatewayFilters);
12   //根据实现 Ordered 优先级接口排序
13   AnnotationAwareOrderComparator.sort(combined);
14   //省略部分代码......
15  
16   //构建默认 WebFilter 过滤器责任链并执行 filter 方法
17   return new DefaultGatewayFilterChain(combined).filter(exchange);
18}

 1//代码位置：DefaultGatewayFilterChain.filter(ServerWebExchange exchange)
 2@Override
 3public Mono<Void> filter(ServerWebExchange exchange) {
 4   return Mono.defer(() -> {
 5      if (this.index < filters.size()) {
 6         GatewayFilter filter = filters.get(this.index);
 7         DefaultGatewayFilterChain chain = new DefaultGatewayFilterChain(this, this.index + 1);
 8         return filter.filter(exchange, chain);
 9      }
10      else {
11         return Mono.empty(); // complete
12      }
13   });
14}

 1//代码位置：HttpClientOperations.onInboundNext(ChannelHandlerContext ctx, Object msg)
 2@Override
 3protected void onInboundNext(ChannelHandlerContext ctx, Object msg) {
 4  if (msg instanceof HttpResponse) {
 5    HttpResponse response = (HttpResponse) msg;
 6    //省略部分代码.....
 7    
 8   //判断是否为最后一个带有标识头的 HttpContent 消息
 9  if (msg instanceof LastHttpContent) {
10    //省略部分代码.....
11    channel().config().setAutoRead(true);
12    if (markSentBody()) {
13      markPersistent(false);
14    }
15    //执行终止逻辑
16    terminate();
17    return;
18  }
19  //省略部分代码.....
20  super.onInboundNext(ctx, msg);
21}

 1//代码位置：HttpClientOperations.afterInboundComplete
 2@Override
 3protected void afterInboundComplete() {
 4  if (redirecting != null) {
 5    listener().onUncaughtException(this, redirecting);
 6  }
 7  else {
 8    //触发 RESPONSE_COMPLETED 状态事件
 9    listener().onStateChange(this, HttpClientState.RESPONSE_COMPLETED);
10  }
11}

 1//代码位置：DefaultPooledConnectionProvider.onStateChange(Connection connection, State newState)
 2@Override
 3public void onStateChange(Connection connection, State newState) {
 4 if (newState == State.DISCONNECTING) {
 5		//省略部分代码......
 6    ConnectionObserver obs = channel.attr(OWNER)
 7                                    .getAndSet(ConnectionObserver.emptyListener());
 8    pooledRef.release()
 9             .subscribe(null,t -> {
10                         //省略部分代码......
11                         onTerminate.tryEmitEmpty();
12                         obs.onStateChange(connection, State.RELEASED);
13                     },() -> {
14                         //省略部分代码......
15                         onTerminate.tryEmitEmpty();
16                         obs.onStateChange(connection, State.RELEASED);
17                     });
18    return;
19 }
20 owner().onStateChange(connection, newState);
21}