SpringBoot之webflux全面解析

Posted on 2022-02-24 by WalkonNet

webflux介紹

Spring Boot 2.0

spring.io 官網有句醒目的話是：

BUILD ANYTHING WITH SPRING BOOT

Spring Boot （Boot 顧名思義，是引導的意思）框架是用於簡化 Spring 應用從搭建到開發的過程。

應用開箱即用，隻要通過一個指令，包括命令行 java -jar 、SpringApplication 應用啟動類、 Spring Boot Maven 插件等，就可以啟動應用瞭。

另外，Spring Boot 強調隻需要很少的配置文件，所以在開發生產級 Spring 應用中，讓開發變得更加高效和簡易。

目前，Spring Boot 版本是 2.x 版本。Spring Boot 包括 WebFlux。

傳統的以SpringMVC為代表的webmvc技術使用的是同步阻塞式IO模型

而Spring WebFlux是一個異步非阻塞式IO模型，可以用少量的容器線程支撐大量的並發訪問，所以Spring WebFlux可以提升吞吐量和伸縮性，但是接口的響應時間並不會縮短，其處理結果還是得由worker線程處理完成之後在返回給請求

webflux應用場景

適合IO密集型、磁盤IO密集、網絡IO密集等服務場景，比如微服務網關，就可以使用webflux技術來顯著的提升網關對下遊服務的吞吐量，spring cloud gateway就使用瞭webflux這門技術

Spring Boot 2.0 WebFlux

瞭解 WebFlux，首先瞭解下什麼是 Reactive Streams。Reactive Streams 是 JVM 中面向流的庫標準和規范：

處理可能無限數量的元素
按順序處理
組件之間異步傳遞
強制性非阻塞背壓（Backpressure）

Backpressure（背壓）

背壓是一種常用策略，使得發佈者擁有無限制的緩沖區存儲元素，用於確保發佈者發佈元素太快時，不會去壓制訂閱者。

Reactive Streams（響應式流）

一般由以下組成：

一般由以下組成：

publisher：發佈者，發佈元素到訂閱者
subscriber：訂閱者，消費元素
subscription：訂閱，在發佈者中，訂閱被創建時，將與訂閱者共享
processor：處理器，發佈者與訂閱者之間處理數據，包含瞭發佈者與訂閱者的共同體

publisher接口規范

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> var1);
}

subscriber接口規范

public interface Subscriber<T> {
    void onSubscribe(Subscription var1); 
    void onNext(T var1); 
    void onError(Throwable var1); 
    void onComplete();
}

subscription接口規范

public interface Subscription {
    void request(long var1); 
    void cancel();
}

processor接口規范

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

響應式編程

有瞭 Reactive Streams 這種標準和規范，利用規范可以進行響應式編程。那再瞭解下什麼是 Reactive programming 響應式編程。響應式編程是基於異步和事件驅動的非阻塞程序，隻是垂直通過在 JVM 內啟動少量線程擴展，而不是水平通過集群擴展。這就是一個編程范例，具體項目中如何體現呢？

響應式項目編程實戰中，通過基於 Reactive Streams 規范實現的框架 Reactor 去實戰。Reactor 一般提供兩種響應式 API ：

Mono：實現發佈者，並返回 0 或 1 個元素
Flux：實現發佈者，並返回 N 個元素

Spring Webflux

Spring Boot Webflux 就是基於 Reactor 實現的。Spring Boot 2.0 包括一個新的 spring-webflux 模塊。該模塊包含對響應式 HTTP 和 WebSocket 客戶端的支持，以及對 REST，HTML 和 WebSocket 交互等程序的支持。一般來說，Spring MVC 用於同步處理，Spring Webflux 用於異步處理。

Spring Boot Webflux 有兩種編程模型實現，一種類似 Spring MVC 註解方式，另一種是使用其功能性端點方式。

Spring Boot 2.0 WebFlux 特性

常用的 Spring Boot 2.0 WebFlux 生產的特性如下：

響應式 API
編程模型
適用性
內嵌容器
Starter 組件

還有對日志、Web、消息、測試及擴展等支持。

響應式 API

Reactor 框架是 Spring Boot Webflux 響應庫依賴，通過 Reactive Streams 並與其他響應庫交互。提供瞭兩種響應式 API：Mono 和 Flux。一般是將 Publisher 作為輸入，在框架內部轉換成 Reactor 類型並處理邏輯，然後返回 Flux 或 Mono 作為輸出。

spring webflux和spring mvc的異同點

一圖就很明確瞭，WebFlux 和 MVC 有交集，方便大傢遷移。但是註意：

MVC 能滿足場景的，就不需要更改為 WebFlux。
要註意容器的支持，可以看看下面內嵌容器的支持。
微服務體系結構，WebFlux 和 MVC 可以混合使用。尤其開發 IO 密集型服務的時候，選擇 WebFlux 去實現。
spring mvc是一個命令式的編程方式采用同步阻塞方式，方便開發人員編寫代碼和調試；spring webflux調試會非常不方便
JDBC連接池和JPA等技術還是阻塞模型，傳統的關系型數據庫如MySQL也不支持非阻塞的方式獲取數據，目前隻有非關系型數據庫如Redis、Mongodb支持非阻塞方式獲取數據

編程模型

Spring 5 web 模塊包含瞭 Spring WebFlux 的 HTTP 抽象。類似 Servlet API , WebFlux 提供瞭 WebHandler API 去定義非阻塞 API 抽象接口。可以選擇以下兩種編程模型實現：

註解控制層。和 MVC 保持一致，WebFlux 也支持響應性 @RequestBody 註解。
功能性端點。基於 lambda 輕量級編程模型，用來路由和處理請求的小工具。和上面最大的區別就是，這種模型，全程控制瞭請求 – 響應的生命流程

內嵌容器

跟 Spring Boot 大框架一樣啟動應用，但 WebFlux 默認是通過 Netty 啟動，並且自動設置瞭默認端口為 8080。另外還提供瞭對 Jetty、Undertow 等容器的支持。開發者自行在添加對應的容器 Starter 組件依賴，即可配置並使用對應內嵌容器實例。

但是要註意，必須是 Servlet 3.1+ 容器，如 Tomcat、Jetty；或者非 Servlet 容器，如 Netty 和 Undertow。

Netty優點

API使用簡單、易上手
功能強大、支持多種主流協議
定制能力強、可擴展性高
性能高、綜合性能最優
成熟穩定、久經考驗
社區活躍、學習資料多

Netty selector模型

Reactor指南

Reactor 框架是 Pivotal 公司（開發 Spring 等技術的公司）開發的
實現瞭 Reactive Programming 思想，符合Reactive Streams 規范（Reactive Streams 是由 Netflix、TypeSafe、Pivotal 等公司發起的）的一項技術
側重於server端的響應式編程框架
Reactor 框架主要有兩個主要的模塊：reactor-core 和 reactor-ipc。前者主要負責 Reactive Programming 相關的核心 API 的實現，後者負責高性能網絡通信的實現，目前是基於 Netty 實現的。

Java原有的異步編程方式

Callback：異步方法采用一個callback作為參數，當結果出來後回調這個callback，例如swings的EventListener
Future：異步方法返回一個Future<T>，此時結果並不是立刻可以拿到，需要處理結束之後才可以使用

Future局限

多個Future組合不易
調用Future#get時仍然會阻塞
缺乏對多個值以及進一步的出錯處理

Reactor的Publisher

Mono 實現瞭 org.reactivestreams.Publisher 接口，代表0到1個元素的響應式序列。
Flux 同樣實現瞭 org.reactivestreams.Publisher 接口，代表0到N個元素的結果。

Flux介紹

Flux<T>是一個標準Publisher<T>，表示0到N個發射項的異步序列，可選地以完成信號或錯誤終止。與Reactive Streams規范中一樣，這三種類型的信號轉換為對下遊訂閱者的onNext、onComplete或onError方法的調用。
在這種大范圍的可能信號中，Flux是通用的reactive 類型。註意，所有事件，甚至終止事件，都是可選的：沒有onNext事件，但是onComplete事件表示一個空的有限序列，但是移除onComplete並且您有一個無限的空序列（除瞭關於取消的測試之外，沒有特別有用）。同樣，無限序列不一定是空的。例如，Flux.interval(Duration) 產生一個Flux<Long>，它是無限的，從時鐘發出規則的數據。

Mono介紹

Mono<T>是一個專門的Publisher<T>，它最多發出一個項，然後可選地以onComplete信號或onError信號結束。
它隻提供瞭可用於Flux的操作符的子集，並且一些操作符（特別是那些將Mono與另一個發佈者組合的操作符）切換到Flux。
例如，Mono#concatWith(Publisher)返回一個Flux ，而Mono#then(Mono)則返回另一個Mono。
註意，Mono可以用於表示隻有完成概念（類似於Runnable）的無值異步進程。若要創建一個，請使用Mono<Void>。

publisher訂閱

reactor實踐

首先maven工廠引入pom

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTest { 
    @Test
    public void testReactor(){
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6);
        Mono<Integer> mono = Mono.just(1);
 
        Integer[] arr = {1,2,3,4,5,6};
        Flux<Integer> flux1 = Flux.fromArray(arr);
 
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6);
        Flux<Integer> flux2 = Flux.fromIterable(list);
 
        Flux<Integer> flux3 = Flux.from(flux);
 
        Flux<Integer> flux4 = Flux.fromStream(Stream.of(1, 2, 3, 4, 5, 6));
 
        flux.subscribe();
 
        flux1.subscribe(System.out::println);
 
        flux2.subscribe(System.out::println,System.err::println);
 
        flux3.subscribe(System.out::println,System.err::println,() -> System.out.println("complete"));
 
        flux4.subscribe(System.out::println,System.err::println,
                () -> System.out.println("complete"),
                subscription -> subscription.request(3)); 
        flux4.subscribe(new DemoSubscriber());
    }
 
    class DemoSubscriber extends BaseSubscriber<Integer>{
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            System.out.println("Subscribe");
            subscription.request(1);
        }
 
        @Override
        protected void hookOnNext(Integer value) {
            if(value == 4){
                //背壓，通知數據源，不要發送數據瞭
                cancel();
            }
            System.out.println(value);
            request(1);
        }
    }
}

Reactor操作符

map – 元素映射為新元素

map操作可以將數據元素進行轉換/映射，得到一個新元素。

flatMap – 元素映射為流

flatMap操作可以將每個數據元素轉換/映射為一個流，然後將這些流合並為一個大的數據流。

filter – 過濾

filter操作可以對數據元素進行篩選。

zip – 一對一合並

看到zip這個詞可能會聯想到拉鏈，它能夠將多個流一對一的合並起來。zip有多個方法變體，我們介紹一個最常見的二合一的。

Reactor中提供瞭非常豐富的操作符，除瞭以上幾個常見的，還有：

用於編程方式自定義生成數據流的create和generate等及其變體方法；
用於“無副作用的peek”場景的doOnNext、doOnError、doOncomplete、doOnSubscribe、doOnCancel等及其變體方法；
用於數據流轉換的when、and/or、merge、concat、collect、count、repeat等及其變體方法；
用於過濾/揀選的take、first、last、sample、skip、limitRequest等及其變體方法；
用於錯誤處理的timeout、onErrorReturn、onErrorResume、doFinally、retryWhen等及其變體方法；
用於分批的window、buffer、group等及其變體方法；
用於線程調度的publishOn和subscribeOn方法。

使用這些操作符，你幾乎可以搭建出能夠進行任何業務需求的數據處理管道/流水線。

抱歉以上這些暫時不能一一介紹，更多詳情請參考JavaDoc

reactor和java8 stream區別

形似而神不似

reactor：push模式，服務端推送數據給客戶端
java8 stream：pull模式，客戶端主動向服務端請求數據

Reactor線程模型

Reactor創建線程的方式

Schedulers.immediate()：當前線程
Schedulers.single()：可重用的單線程，註意，這個方法對所有調用者都提供同一個線程來使用，直到該調度器被廢棄。如果你想使用獨占的線程，請使用Schedulers.newSingle()；
Schedulers.elastic()：彈性線程池，它根據需要創建一個線程池，重用空閑線程。線程池如果空閑時間過長（默認為 60s）就會被廢棄。對於 I/O 阻塞的場景比較適用。Schedulers.elastic()能夠方便地給一個阻塞的任務分配它自己的線程，從而不會妨礙其他任務和資源；
Schedulers.parallel()：固定大小線程池，所創建線程池的大小與CPU個數等同
Schedulers.fromExecutorService(ExecutorService)：自定義線程池，基於自定義的ExecutorService創建 Scheduler（雖然不太建議，不過你也可以使用Executor來創建）

線程模型

線程切換實踐

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTest { 
    @Test
    public void testReactor() throws InterruptedException {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6);
 
        flux.map(i -> {
            System.out.println(Thread.currentThread().getName()+"-map1");
            return i * 3;
        }).publishOn(Schedulers.elastic()).map(
                i -> {
                    System.out.println(Thread.currentThread().getName()+"-map2");
                    return i / 3;
                }
        ).subscribeOn(Schedulers.parallel())
                .subscribe(i -> System.out.println(Thread.currentThread().getName()+"-" + i)); 
        Thread.sleep(10000);
    }
}

線程切換總結

publishOn：它將上遊信號傳給下遊，同時改變後續的操作符的執行所在線程，直到下一個publishOn出現在這個鏈上
subscribeOn：作用於向上的訂閱鏈，無論處於操作鏈的什麼位置，它都會影響到源頭的線程執行環境，但不會影響到後續的publishOn

webflux實踐

兼容spring mvc的寫法

@RestController
public class DemoController { 
    @GetMapping("/demo")
    public Mono<String> demo(){
        return Mono.just("demo");
    }
}

spring webflux函數式寫法

@Component
public class DemoHandler { 
    public Mono<ServerResponse> hello(ServerRequest request){
        return ok().contentType(MediaType.TEXT_PLAIN)
                .body(Mono.just("hello"),String.class);
    }
 
    public Mono<ServerResponse> world(ServerRequest request){
        return ok().contentType(MediaType.TEXT_PLAIN)
                .body(Mono.just("world"),String.class);
    }
 
    public Mono<ServerResponse> times(ServerRequest request){
        //每隔一秒發送當前的時間
        return ok().contentType(MediaType.TEXT_EVENT_STREAM)
                .body(Flux.interval(Duration.ofSeconds(1))
                        .map(it -> new SimpleDateFormat("HH:mm:ss").format(new Date())),String.class);
    }
}

配置路由

@Configuration
public class RouterConfig { 
    @Autowired
    private DemoHandler demoHandler;
 
    @Bean
    public RouterFunction<ServerResponse> demoRouter(){
        //路由函數的編寫
        return route(GET("/hello"),demoHandler::hello)
                .andRoute(GET("/world"),demoHandler::world)
                .andRoute(GET("/times"),demoHandler::times);
    }
}

連接關系型數據庫案例

@Component
public class DemoHandler { 
    @Autowired
    private PersonService personService; 
    public Mono<ServerResponse> queryPerson(ServerRequest request){
        Integer id = Integer.valueOf(request.pathVariable("id"));
        return ok().contentType(MediaType.APPLICATION_JSON_UTF8)
                .body(Mono.just(personService.getPersonById(id)), Person.class);
    }
}

配置路由

@Configuration
public class RouterConfig { 
    @Autowired
    private DemoHandler demoHandler;
 
    @Bean
    public RouterFunction<ServerResponse> demoRouter(){
        //路由函數的編寫
        return route(GET("/hello"),demoHandler::hello)
                .andRoute(GET("/world"),demoHandler::world)
                .andRoute(GET("/times"),demoHandler::times)
                .andRoute(GET("/queryPerson/{id}"),demoHandler::queryPerson);
    }
}

連接非關系型數據庫案例

引入mongodb的maven

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>

在application.properties中配置mongodb屬性

#mongodb
spring.data.mongodb.uri=mongodb://root:yibo@localhost:27017
spring.data.mongodb.database=webflux

編寫代碼

@Document(collection = "user")
@Data
public class User { 
    @Id
    private String id; 
    private String name; 
    private int age;
}
 
@Repository
public interface UserRepository extends ReactiveMongoRepository<User,String> {
}
 
@Component
public class DemoHandler { 
    @Autowired
    private UserRepository userRepository; 
    public Mono<ServerResponse> listUser(ServerRequest request){
        return ok().contentType(MediaType.APPLICATION_JSON_UTF8)
                .body(userRepository.findAll(), User.class);
    }
 
    public Mono<ServerResponse> saveUser(ServerRequest request){
        String name = request.pathVariable("name");
        Integer age = Integer.valueOf(request.pathVariable("age"));
        User user = new User();
        user.setName(name);
        user.setAge(age);
        Mono<User> mono = Mono.just(user);
        return ok().build(userRepository.insert(mono).then());
    }
}

編寫路由

@Configuration
public class RouterConfig { 
    @Autowired
    private DemoHandler demoHandler;
 
    @Bean
    public RouterFunction<ServerResponse> demoRouter(){
        //路由函數的編寫
        return route(GET("/hello"),demoHandler::hello)
                .andRoute(GET("/world"),demoHandler::world)
                .andRoute(GET("/times"),demoHandler::times)
                .andRoute(GET("/queryPerson/{id}"),demoHandler::queryPerson)
                .andRoute(GET("/listUser"),demoHandler::listUser)
                .andRoute(GET("/saveUser/{name}/{age}"),demoHandler::saveUser);
    }
}

webflux解析

spring mvc處理流程

具體步驟：

第一步：發起請求到前端控制器(DispatcherServlet)
第二步：前端控制器請求HandlerMapping查找 Handler （可以根據xml配置、註解進行查找）
匹配條件包括：請求路徑、請求方法、header信息等
第三步：處理器映射器HandlerMapping向前端控制器返回Handler，HandlerMapping會把請求映射為HandlerExecutionChain對象（包含一個Handler處理器（頁面控制器）對象，多個HandlerInterceptor攔截器對象），通過這種策略模式，很容易添加新的映射策略
HandlerInterceptor是請求路徑上的攔截器，需要自己實現這個接口以攔截請求，做一些對handler的前置和後置處理工作。
第四步：前端控制器調用處理器適配器去執行Handler
第五步：處理器適配器HandlerAdapter將會根據適配的結果去執行Handler
第六步：Handler執行完成給適配器返回ModelAndView
第七步：處理器適配器向前端控制器返回ModelAndView （ModelAndView是springmvc框架的一個底層對象，包括 Model和view）
第八步：前端控制器請求視圖解析器去進行視圖解析（根據邏輯視圖名解析成真正的視圖(jsp)），通過這種策略很容易更換其他視圖技術，隻需要更改視圖解析器即可
第九步：視圖解析器向前端控制器返回View
第十步：前端控制器進行視圖渲染（視圖渲染將模型數據(在ModelAndView對象中)填充到request域）
第十一步：前端控制器向用戶響應結果

spring webflux處理請求流程

核心控制器DispatcherHandler，等同於阻塞方式的DispatcherServlet

DispatcherHandler實現ApplicationContextAware，那麼必然會調用setApplicationContext方法

public class DispatcherHandler implements WebHandler, ApplicationContextAware {
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) {
        initStrategies(applicationContext);
    }
}

initStrategies初始化

獲取HandlerMapping，HandlerAdapter，HandlerResultHandler的所有實例

protected void initStrategies(ApplicationContext context) {
    //獲取HandlerMapping及其子類型的bean
    //HandlerMapping根據請求request獲取handler執行鏈
    Map<String, HandlerMapping> mappingBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
            context, HandlerMapping.class, true, false);
 
    ArrayList<HandlerMapping> mappings = new ArrayList<>(mappingBeans.values());
    //排序
    AnnotationAwareOrderComparator.sort(mappings);
    this.handlerMappings = Collections.unmodifiableList(mappings);
 
    //獲取HandlerAdapter及其子類型的bean
    Map<String, HandlerAdapter> adapterBeans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
            context, HandlerAdapter.class, true, false);
 
    this.handlerAdapters = new ArrayList<>(adapterBeans.values());
    //排序
    AnnotationAwareOrderComparator.sort(this.handlerAdapters);
 
    //獲取HandlerResultHandler及其子類型的bean
    Map<String, HandlerResultHandler> beans = BeanFactoryUtils.beansOfTypeIncludingAncestors(
            context, HandlerResultHandler.class, true, false);
 
    this.resultHandlers = new ArrayList<>(beans.values());
    AnnotationAwareOrderComparator.sort(this.resultHandlers);
}

webflux中引入瞭一個新的HandlerMapping，即RouterFunctionMapping

RouterFunctionMapping實現瞭InitializingBean，因此在其實例化的時候，會調用afterPropertiesSet方法

public class RouterFunctionMapping extends AbstractHandlerMapping implements InitializingBean {
 
    @Nullable
    private RouterFunction<?> routerFunction;
 
    //讀取http傳輸數據，並解碼成一個對象
    private List<HttpMessageReader<?>> messageReaders = Collections.emptyList();
 
    public RouterFunctionMapping(RouterFunction<?> routerFunction) {
        this.routerFunction = routerFunction;
    }
 
    @Nullable
    public RouterFunction<?> getRouterFunction() {
        return this.routerFunction;
    }
 
    public void setMessageReaders(List<HttpMessageReader<?>> messageReaders) {
        this.messageReaders = messageReaders;
    }
 
    @Override
    public void afterPropertiesSet() throws Exception {
        if (CollectionUtils.isEmpty(this.messageReaders)) {
            ServerCodecConfigurer codecConfigurer = ServerCodecConfigurer.create();
            this.messageReaders = codecConfigurer.getReaders();
        }
 
        if (this.routerFunction == null) {
            //afterPropertiesSet方法調用的時候，routerFunction為null
            initRouterFunctions();
        }
    }
 
    protected void initRouterFunctions() {
        //獲取routerFunctions集合
        List<RouterFunction<?>> routerFunctions = routerFunctions();
         //將一個請求中含有多個路由請求RouterFunction合並成一個RouterFunction
        this.routerFunction = routerFunctions.stream().reduce(RouterFunction::andOther).orElse(null);
        logRouterFunctions(routerFunctions);
    }
 
    private List<RouterFunction<?>> routerFunctions() {
        //obtainApplicationContext()獲取ApplicationContext對象
        List<RouterFunction<?>> functions = obtainApplicationContext()
                //獲取指定bean的提供者，即上文配置的路由類
                .getBeanProvider(RouterFunction.class)
                //排序
                .orderedStream()
                //將流裡面的都強轉成RouterFunction對象
                .map(router -> (RouterFunction<?>)router)
                .collect(Collectors.toList());
        return (!CollectionUtils.isEmpty(functions) ? functions : Collections.emptyList());
    }
 
    private void logRouterFunctions(List<RouterFunction<?>> routerFunctions) {
        //判斷當前的日志級別是否是Debug
        if (logger.isDebugEnabled()) {
            int total = routerFunctions.size();
            String message = total + " RouterFunction(s) in " + formatMappingName();
            if (logger.isTraceEnabled()) {
                if (total > 0) {
                    routerFunctions.forEach(routerFunction -> logger.trace("Mapped " + routerFunction));
                }
                else {
                    logger.trace(message);
                }
            }
            else if (total > 0) {
                logger.debug(message);
            }
        }
    }
    ......
}

webflux中引入瞭一個新的HandlerAdapter，即HandlerFunctionAdapter
webflux中引入瞭一個新的HandlerResultHandler，即ServerResponseResultHandler

ServerResponseResultHandler實現瞭InitializingBean，因此在其實例化的時候，會調用afterPropertiesSet方法

流式處理請求handler()

@Override
public Mono<Void> handle(ServerWebExchange exchange) {
    //handlerMappings在initStrategies()方法中已經構造好瞭
    if (this.handlerMappings == null) {
        return createNotFoundError();
    }
    //構造Flux，數據源為handlerMappings集合
    return Flux.fromIterable(this.handlerMappings)
            //獲取Mono<Handler>對象，通過concatMap保證順序和handlerMappings順序一致
            //嚴格保證順序是因為在一個系統中可能存在一個Url有多個能夠處理的HandlerMapping的情況
            .concatMap(mapping -> mapping.getHandler(exchange))
            .next()
            //如果next()娶不到值則拋出錯誤
            .switchIfEmpty(createNotFoundError())
            //觸發HandlerApter的handle方法
            .flatMap(handler -> invokeHandler(exchange, handler))
            //觸發HandlerResultHandler 的handleResult方法
            .flatMap(result -> handleResult(exchange, result));
}

觸發HandlerApter的handle方法

private Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
    return getResultHandler(result).handleResult(exchange, result)
            .onErrorResume(ex -> result.applyExceptionHandler(ex).flatMap(exceptionResult ->
                    getResultHandler(exceptionResult).handleResult(exchange, exceptionResult)));
}
 
private HandlerResultHandler getResultHandler(HandlerResult handlerResult) {
    if (this.resultHandlers != null) {
        for (HandlerResultHandler resultHandler : this.resultHandlers) {
            if (resultHandler.supports(handlerResult)) {
                return resultHandler;
            }
        }
    }
    throw new IllegalStateException("No HandlerResultHandler for " + handlerResult.getReturnValue());
}

總結

DispatcherHandler的流程是

1、通過 HandlerMapping（和DispathcherServlet中的HandlerMapping不同）獲取到HandlerAdapter放到ServerWebExchange的屬性中

2、獲取到HandlerAdapter後觸發handle方法，得到HandlerResult3、通過HandlerResult，觸發handleResult，針對不同的返回類找到不同的HandlerResultHandler如視圖渲染ViewResolutionResultHandler、ServerResponseResultHandler、ResponseBodyResultHandler、ResponseEntityResultHandler不同容器有不同的實現，如Reactor，Jetty，Tomcat等。

以上為個人經驗，希望能給大傢一個參考，也希望大傢多多支持WalkonNet。

SpringBoot之webflux全面解析

目錄

webflux介紹

webflux應用場景

Spring Boot 2.0 WebFlux

響應式編程

Spring Webflux

spring webflux和spring mvc的異同點

Netty selector模型

Reactor指南

Java原有的異步編程方式

Reactor線程模型

webflux實踐

webflux解析

總結

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

webflux介紹

webflux應用場景

Spring Boot 2.0 WebFlux

響應式編程

Spring Webflux

spring webflux和spring mvc的異同點

Netty selector模型

Reactor指南

Java原有的異步編程方式

Reactor線程模型

webflux實踐

webflux解析

總結

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆