欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

spring?webflux響應(yīng)式編程使用詳解

 更新時(shí)間：2023年10月27日 09:47:35   作者：yysx  
webflux,即響應(yīng)式編程,響應(yīng)式編程是一種用于處理異步數(shù)據(jù)流和事件的編程范式,spring?webflux是spring在5.0版本后提供的一套響應(yīng)式編程風(fēng)格的web開(kāi)發(fā)框架,本文給大家詳細(xì)講講spring?webflux響應(yīng)式編程的使用,需要的朋友可以參考下

一 、響應(yīng)式編程

響應(yīng)式編程（reactive programming）是一種基于數(shù)據(jù)流（data stream）和變化傳遞（propagation of change）的聲明式（declarative）的編程范式

（1）數(shù)據(jù)流

將要處理的數(shù)據(jù)抽象成流

通用的流式處理：

生產(chǎn)者產(chǎn)生數(shù)據(jù)，對(duì)數(shù)據(jù)進(jìn)行中間處理，消費(fèi)者拿到數(shù)據(jù)消費(fèi)。

（2）變化傳遞

● 在命令式編程(我們的日常編程模式)下，式子a=b+c，這就意味著a的值是由b和c計(jì)算出來(lái)的。如果b或者c后續(xù)有變化，不會(huì)影響到a的值。

● 在響應(yīng)式編程下，式子a:=b+c，這就意味著a的值是由b和c計(jì)算出來(lái)的。但如果b或者c的值后續(xù)有變化，會(huì)影響到a的值。

（3）聲明式

通過(guò)API來(lái)處理數(shù)據(jù)流中的數(shù)據(jù)。

適用場(chǎng)景：異步非阻塞。響應(yīng)式系統(tǒng)具有某些特性，使其成為低延遲、高吞吐量工作負(fù)載的理想選擇。

比如：

一個(gè)日志監(jiān)控系統(tǒng)，我們的前端頁(yè)面將不再需要通過(guò)“命令式”的輪詢的方式不斷向服務(wù)器請(qǐng)求數(shù)據(jù)然后進(jìn)行更新，而是在建立好通道之后，數(shù)據(jù)流從系統(tǒng)源源不斷流向頁(yè)面，從而展現(xiàn)實(shí)時(shí)的指標(biāo)變化曲線；

一個(gè)社交平臺(tái)，朋友的動(dòng)態(tài)、點(diǎn)贊和留言不是手動(dòng)刷出來(lái)的，而是當(dāng)后臺(tái)數(shù)據(jù)變化的時(shí)候自動(dòng)體現(xiàn)到界面上的。

二 、響應(yīng)式流

響應(yīng)式流(Reactive Streams)通過(guò)定義一組實(shí)體，接口和互操作方法，給出了實(shí)現(xiàn)異步非阻塞背壓的標(biāo)準(zhǔn)。第三方遵循這個(gè)標(biāo)準(zhǔn)來(lái)實(shí)現(xiàn)具體的解決方案，常見(jiàn)的有Reactor，RxJava，Akka Streams，Ratpack等。

常見(jiàn)的響應(yīng)流庫(kù)簡(jiǎn)介

（1）JDK9響應(yīng)式流：

Publisher：發(fā)布者，數(shù)據(jù)來(lái)源，又稱生產(chǎn)者。

Subscriber：訂閱者，數(shù)據(jù)目的地，又稱消費(fèi)者。

Processor：發(fā)布者和訂閱者之間處理數(shù)據(jù)。

（2）Reactor響應(yīng)式流庫(kù)

發(fā)布者：Mono（返回0或1個(gè)元素）、Flux（返回0~n個(gè)元素）。

訂閱者：Spring框架來(lái)完成。

背壓：

Backpressure

在異步場(chǎng)景中，被觀察者發(fā)送事件的速度遠(yuǎn)遠(yuǎn)快于觀察者的處理速度情況下，一種告知上游的被觀察者降低發(fā)送速度的策略。

生產(chǎn)者和消費(fèi)者模式下，如果生產(chǎn)者生產(chǎn)的消費(fèi)多了，消費(fèi)者消費(fèi)不過(guò)來(lái)，就容易被壓垮了。

借助背壓，消費(fèi)者可以告訴生產(chǎn)者自己需要多少量的數(shù)據(jù)。

響應(yīng)式流=異步非阻塞+流量控制。

三、Spring WebFlux

Spring提供了兩條并行的技術(shù)棧，一條是基于Servlet API的Spring MVC和Spring Data，另一個(gè)條是完整的響應(yīng)式棧，包括Spring WebFlux和Spring Data’s reactive repositories。

在這里插入圖片描述

SpringBoot提供了自動(dòng)配置的Spring Webflux，簡(jiǎn)化了響應(yīng)式web應(yīng)用的開(kāi)發(fā)。

1、整合Webflux

pom引入依賴：

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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

啟動(dòng)類(lèi)添加@EnableWebFlux。

兩種方式：

（1）基于注解

@RestController
@RequestMapping("/user")
public class MyController {
    
    @GetMapping("/list")
    public Flux<User> getList(){
        return Flux.fromIterable(list.stream().filter(k->{
           return k.getId()>3;
        }).collect(Collectors.toList()));
    }
    
    @GetMapping("/{id}")
    public Mono<User> getUser(@PathVariable Long userId) {
        return this.userRepository.findById(userId);
    }
}

（2）函數(shù)式

使用handler代替controller，做流的處理。

引入路由，類(lèi)似SpringMVC中的@RequestMapping。

@Component
public class UserHandler {

    @Resource
    UserServiceImpl userService;
    
    public Mono<ServerResponse> getList(ServerRequest serverRequest){
        Flux<User> userFlux=userService.getList();
        return ServerResponse.ok().body(userFlux, User.class);
    }
}

@Configuration(proxyBeanMethods = false)
public class UserRouting {

    @Bean
    public RouterFunction<ServerResponse> monoRouterFunction(UserHandler userHandler){
        return RouterFunctions
                .route(GET("/user/list"),userHandler::getList)
                .andRoute(GET("/hello").and(accept(TEXT_PLAIN)),userHandler::hello)
                .andRoute(GET("/hello/stream").and(accept(APPLICATION_STREAM_JSON)),userHandler::helloStream);
    }
}

啟動(dòng)后，可以看到使用的是內(nèi)置的netty。

訪問(wèn)http://localhost:20001/user/list 即可。

   @GetMapping("list/getStringSleep")
    public Mono<String> getStringSleep() {
        log.info("invoke start");
        Mono<String> s=Mono.fromSupplier(()->createStr());
        log.info("invoke end");
        return s;
    }

   private String createStr() {
        try {
            TimeUnit.SECONDS.sleep(5);
            System.out.println("sleep");
        } catch (InterruptedException e) {
        }
        return "some string";
    }

先執(zhí)行invoke start invoke end 再執(zhí)行異步流處理中的sleep日志。

2022-02-18 13:47:26.364  INFO 61992 --- [ctor-http-nio-2] com.example.controller.MyController      : invoke start
2022-02-18 13:47:26.366  INFO 61992 --- [ctor-http-nio-2] com.example.controller.MyController      : invoke end
2022-02-18 13:47:26.369 DEBUG 61992 --- [ctor-http-nio-2] o.s.w.r.r.m.a.ResponseBodyResultHandler  : [fceca854-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:53707] Using 'text/html' given [text/html, application/xhtml+xml, image/avif, image/webp, image/apng, application/xml;q=0.9, application/signed-exchange;v=b3;q=0.9, */*;q=0.8] and supported [text/plain;charset=UTF-8, text/event-stream, text/plain;charset=UTF-8, */*]
2022-02-18 13:47:26.369 DEBUG 61992 --- [ctor-http-nio-2] o.s.w.r.r.m.a.ResponseBodyResultHandler  : [fceca854-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:53707] 0..1 [java.lang.String]
sleep
2022-02-18 13:47:31.375 TRACE 61992 --- [ctor-http-nio-2] o.s.core.codec.CharSequenceEncoder       : [fceca854-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:53707] Writing "some string"
2022-02-18 13:47:31.376 DEBUG 61992 --- [ctor-http-nio-2] r.n.http.server.HttpServerOperations     : [fceca854-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:53707] Decreasing pending responses, now 0
2022-02-18 13:47:31.377 DEBUG 61992 --- [ctor-http-nio-2] r.n.http.server.HttpServerOperations     : [fceca854-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:53707] Last HTTP packet was sent, terminating the channel
2022-02-18 13:47:31.377 TRACE 61992 --- [ctor-http-nio-2] o.s.w.s.adapter.HttpWebHandlerAdapter    : [fceca854-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:53707] Completed 200 OK, headers={masked}
2022-02-18 13:47:31.377 TRACE 61992 --- [ctor-http-nio-2] o.s.h.s.r.ReactorHttpHandlerAdapter      : [fceca854-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:53707] Handling completed
2022-02-18 13:47:31.377 DEBUG 61992 --- [ctor-http-nio-2] r.n.http.server.HttpServerOperations     : [fceca854-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:53707] Last HTTP response frame
2022-02-18 13:47:31.378 TRACE 61992 --- [ctor-http-nio-2] reactor.netty.channel.ChannelOperations  : [fceca854, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:53707] Disposing ChannelOperation from a channel

2、事件推送

Server-Sent Event(SSEs)

 @GetMapping(value = "push",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> push() {
        log.info("push start");
        Flux<String> result= Flux.fromStream(IntStream.range(1,5).mapToObj(i->{
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
            }
            return "push data--" + i;
        }));
        log.info("push end");
        return  result;
    }

在瀏覽器中執(zhí)行http://localhost:8080/demo/push，在服務(wù)器終端可以看到日志，已經(jīng)執(zhí)行完成。但此時(shí)頁(yè)面上的推送還在進(jìn)行中。

在這里插入圖片描述

從瀏覽器的監(jiān)控可以看到，瀏覽器接收的是一個(gè)事件流 EventStream。

在這里插入圖片描述

3、實(shí)現(xiàn)背壓

WebFlux中可以用limitRate()設(shè)置背壓的限制參數(shù)。

測(cè)試代碼：

其中l(wèi)ist為內(nèi)存中構(gòu)造的數(shù)據(jù)集合，此處忽略。

//不使用背壓 
@GetMapping(value = "list")
    public Flux<User> getList() {
        log.info("invoke start");
        Flux<User> user=Flux.fromStream(list.stream());
        log.info("invoke end");
        return user;
    }

//使用背壓
    @GetMapping(value = "list/setLimit",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<User> getListLimit() {
        log.info("invoke start");
        Flux<User> user=Flux.fromStream(list.stream()).limitRate(2);
        log.info("invoke end");
        return user;
    }

對(duì)比使用背壓和不適用背壓：

● 接口的總耗時(shí)

在這里插入圖片描述

● 詳細(xì)耗時(shí)

使用背壓：

在這里插入圖片描述

不使用背壓：

在這里插入圖片描述

瀏覽器接收的數(shù)據(jù)格式略有差異：

不使用背壓：
[{“id”:0,“name”:“user0”},{“id”:1,“name”:“user1”},{“id”:2,“name”:“user2”},{“id”:3,“name”:“user3”},{“id”:4,“name”:“user4”},{“id”:5,“name”:“user5”}]

使用背壓：

data:{“id”:0,“name”:“user0”}

data:{“id”:1,“name”:“user1”}

data:{“id”:2,“name”:“user2”}

data:{“id”:3,“name”:“user3”}

data:{“id”:4,“name”:“user4”}

data:{“id”:5,“name”:“user5”}

四、配置數(shù)據(jù)源（以mysql為例）

我們最常使用的 JDBC 其實(shí)是同步的，而我們使用 WebFlux 是為了通過(guò)異步的方式來(lái)提高服務(wù)端的響應(yīng)效率，WebFlux 雖然實(shí)現(xiàn)了異步，但是由于 JDBC 還是同步的，而大部分應(yīng)用都是離不開(kāi)數(shù)據(jù)庫(kù)的，所以其實(shí)效率本質(zhì)上還是沒(méi)有提升。

常見(jiàn)的異步JDBC有：ADAB、R2DBC。

下面以R2DBC為例，介紹webflux配置mysql。

1、CRUD

● pom引入

   <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-data-r2dbc</artifactId>
    </dependency>
    <dependency>
      <groupId>dev.miku</groupId>
      <artifactId>r2dbc-mysql</artifactId>
    </dependency>

● yaml配置數(shù)據(jù)源地址

當(dāng)然，使用java config的方式配置也可以。

spring:
  r2dbc:
    url: r2dbcs:mysql://localhost:3306/database?sslMode=DISABLED
    username: 
    password:

注意，這里url中的數(shù)據(jù)庫(kù)連接協(xié)議不在是jdbc，而是r2dbcs

● 操作數(shù)據(jù)

這里有兩種方式，都可以實(shí)現(xiàn)。

//方法一：通過(guò)ReactiveCrudRepository

//類(lèi)似DAO
public interface UserRepository extends ReactiveCrudRepository<User,Long> {
}

//在Controller中注入，并使用。

@Resource
UserRepository userRepository;

@GetMapping("/list")
public Flux<User> getListFromDB(){
     return userRepository.findAll();
}

//方法二：使用 DatabaseClient

@Resource
DatabaseClient databaseClient;

@GetMapping("/{id}")
public Mono<Map<String,Object>> getDetail(@PathVariable("id") Integer id){
       return databaseClient.sql("select * from user where id="+id).fetch().one();
}

2、R2DBC

Reactive Relational Database Connectivity，是一個(gè)使用響應(yīng)式驅(qū)動(dòng)集成關(guān)系型數(shù)據(jù)庫(kù)的孵化器。Spring Data R2DBC運(yùn)用熟悉的Spring抽象和repository支持R2DBC。

（1）如何體現(xiàn)異步？

測(cè)試代碼：

		@GetMapping("/{id}")
    public Mono<Map<String,Object>> getDetail(@PathVariable("id") Integer id){
        log.info("invoke start");
        Mono<Map<String,Object>> map= databaseClient.sql("select * from user where id="+id).fetch().one();
        log.info("invoke end");
        return map;
    }

瀏覽器請(qǐng)求 http://localhost:20001/demo/1，從下面的日志可以看出，日志中invoke start、invoke end相繼執(zhí)行，后續(xù)才建立連接，進(jìn)行數(shù)據(jù)庫(kù)查詢操作。


2022-02-11 10:39:36.108  INFO 15483 --- [ctor-http-nio-2] com.example.controller.MyController      : invoke start
2022-02-11 10:39:36.108  INFO 15483 --- [ctor-http-nio-2] com.example.controller.MyController      : invoke end
2022-02-11 10:39:36.109 DEBUG 15483 --- [ctor-http-nio-2] io.r2dbc.pool.ConnectionPool             : Obtaining new connection from the driver
2022-02-11 10:39:36.110 DEBUG 15483 --- [ctor-http-nio-2] o.s.r2dbc.core.DefaultDatabaseClient     : Executing SQL statement [select * from user where id=1]
2022-02-11 10:39:36.110 TRACE 15483 --- [ctor-http-nio-2] o.s.r2dbc.core.DefaultDatabaseClient     : Expanded SQL [select * from user where id=1]
2022-02-11 10:39:36.111 DEBUG 15483 --- [ctor-http-nio-2] dev.miku.r2dbc.mysql.MySqlConnection     : Create a simple statement provided by text query
2022-02-11 10:39:36.111 DEBUG 15483 --- [ctor-http-nio-2] d.m.r.mysql.client.ReactorNettyClient    : Request: SimpleQueryMessage{sql=REDACTED}
2022-02-11 10:39:36.112 TRACE 15483 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : [id: 0x42250dba, L:/127.0.0.1:49383 - R:localhost/127.0.0.1:3306] READ: 121B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 00 00 01 02 28 00 00 02 03 64 65 66 06 66 61 |.....(....def.fa|
|00000010| 6d 69 6c 79 04 75 73 65 72 04 75 73 65 72 02 69 |mily.user.user.i|
|00000020| 64 02 69 64 0c 3f 00 0b 00 00 00 03 00 00 00 00 |d.id.?..........|
|00000030| 00 2c 00 00 03 03 64 65 66 06 66 61 6d 69 6c 79 |.,....def.family|
|00000040| 04 75 73 65 72 04 75 73 65 72 04 6e 61 6d 65 04 |.user.user.name.|
|00000050| 6e 61 6d 65 0c 2d 00 90 01 00 00 fd 00 00 00 00 |name.-..........|
|00000060| 00 09 00 00 04 01 31 06 e5 bc a0 e4 b8 89 07 00 |......1.........|
|00000070| 00 05 fe 00 00 22 00 00 00                      |....."...       |
+--------+-------------------------------------------------+----------------+
2022-02-11 10:39:36.113 DEBUG 15483 --- [actor-tcp-nio-2] d.m.r.mysql.client.MessageDuplexCodec    : Decode context change to DecodeContext-Result
2022-02-11 10:39:36.113 DEBUG 15483 --- [actor-tcp-nio-2] d.m.r.m.m.server.MetadataDecodeContext   : Respond a metadata bundle by filled-up
2022-02-11 10:39:36.113 DEBUG 15483 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : Response: SyntheticMetadataMessage{completed=false, messages=[DefinitionMetadataMessage{database='family', table='user' (origin:'user'), column='id' (origin:'id'), collationId=63, size=11, type=3, definitions=0, decimals=0}, DefinitionMetadataMessage{database='family', table='user' (origin:'user'), column='name' (origin:'name'), collationId=45, size=400, type=253, definitions=0, decimals=0}], eof=null}
2022-02-11 10:39:36.113 DEBUG 15483 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : Response: RowMessage(encoded)
2022-02-11 10:39:36.113 DEBUG 15483 --- [actor-tcp-nio-2] d.m.r.mysql.client.MessageDuplexCodec    : Decode context change to DecodeContext-Command
2022-02-11 10:39:36.114 DEBUG 15483 --- [actor-tcp-nio-2] io.r2dbc.pool.PooledConnection           : Releasing connection
2022-02-11 10:39:36.114 TRACE 15483 --- [actor-tcp-nio-2] o.s.http.codec.json.Jackson2JsonEncoder  : [8cf12dd9-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:49377] Encoding [{id=1, name=張三}]
2022-02-11 10:39:36.115 TRACE 15483 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : [id: 0x42250dba, L:/127.0.0.1:49383 - R:localhost/127.0.0.1:3306] READ COMPLETE
2022-02-11 10:39:36.115 DEBUG 15483 --- [ctor-http-nio-2] r.n.http.server.HttpServerOperations     : [8cf12dd9-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:49377] Decreasing pending responses, now 0
2022-02-11 10:39:36.116 TRACE 15483 --- [ctor-http-nio-2] o.s.w.s.adapter.HttpWebHandlerAdapter    : [8cf12dd9-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:49377] Completed 200 OK, headers={masked}
2022-02-11 10:39:36.116 TRACE 15483 --- [ctor-http-nio-2] o.s.h.s.r.ReactorHttpHandlerAdapter      : [8cf12dd9-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:49377] Handling completed
2022-02-11 10:39:36.116 DEBUG 15483 --- [ctor-http-nio-2] r.n.http.server.HttpServerOperations     : [8cf12dd9-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:49377] Last HTTP response frame
2022-02-11 10:39:36.116 DEBUG 15483 --- [ctor-http-nio-2] r.n.http.server.HttpServerOperations     : [8cf12dd9-2, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:49377] Last HTTP packet was sent, terminating the channel
2022-02-11 10:39:36.116 TRACE 15483 --- [ctor-http-nio-2] reactor.netty.channel.ChannelOperations  : [8cf12dd9, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:49377] Disposing ChannelOperation from a channel

（2）如何使用事務(wù)？

非響應(yīng)式事務(wù)：
代碼在同一線程上執(zhí)行，默認(rèn)同步的方式實(shí)現(xiàn)。采用ThreadLocal的方式，保存連接和會(huì)話信息。

響應(yīng)式事務(wù)：
代碼在不同線程上執(zhí)行，使用Context來(lái)代替ThreadLocal。
類(lèi)似一個(gè)不可變的Map<Object, Object>，采用CopyOnWrite策略，綁定在每一個(gè)訂閱者上。

兩種使用方式：
聲明式事務(wù)

Spring 5.2后，已經(jīng)支持響應(yīng)式事務(wù)，可以直接使用@Transactional注解。

   @Resource
    private DatabaseClient db;
    @Override
    @Transactional
    public Mono<Void> insertRows() {     
        return db.sql("INSERT INTO user (id,name) VALUES(110, 'aaa')")
                .fetch().rowsUpdated()
                .then(db.sql("INSERT INTO user (id,name) VALUES(111, 'aaa')")
                        .then());
    }

主要執(zhí)行日志：


2022-02-14 10:59:06.077 DEBUG 22687 --- [ctor-http-nio-2] o.s.r.c.R2dbcTransactionManager          : Creating new transaction with name [com.example.service.UserServiceImpl.insertRows]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
2022-02-14 10:59:06.077 DEBUG 22687 --- [ctor-http-nio-2] io.r2dbc.pool.ConnectionPool             : Obtaining new connection from the driver
2022-02-14 10:59:06.078 DEBUG 22687 --- [ctor-http-nio-2] o.s.r.c.R2dbcTransactionManager          : Acquired Connection [MonoRetry] for R2DBC transaction
2022-02-14 10:59:06.078 DEBUG 22687 --- [ctor-http-nio-2] o.s.r.c.R2dbcTransactionManager          : Switching R2DBC Connection [PooledConnection[dev.miku.r2dbc.mysql.MySqlConnection@4f99ad97]] to manual commit
2022-02-14 10:59:06.078 DEBUG 22687 --- [ctor-http-nio-2] d.m.r.mysql.client.ReactorNettyClient    : Request: SimpleQueryMessage{sql=REDACTED}
		+-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 53 45 54 20 61 75 74 6f 63 6f 6d 6d 69 74 3d |.SET autocommit=|
|00000010| 30                                              |0               |
+--------+-------------------------------------------------+----------------+


		+-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 42 45 47 49 4e                               |.BEGIN          |
+--------+-------------------------------------------------+----------------+
2022-02-14 10:59:06.080 TRACE 22687 --- [actor-tcp-nio-2] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.example.service.UserServiceImpl.insertRows]
2022-02-14 10:59:06.081 DEBUG 22687 --- [actor-tcp-nio-2] o.s.r2dbc.core.DefaultDatabaseClient     : Executing SQL statement [INSERT INTO user (id,name) VALUES(110, 'aaa')]
2022-02-14 10:59:06.081 TRACE 22687 --- [actor-tcp-nio-2] o.s.r2dbc.core.DefaultDatabaseClient     : Expanded SQL [INSERT INTO user (id,name) VALUES(110, 'aaa')]
+-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 49 4e 53 45 52 54 20 49 4e 54 4f 20 75 73 65 |.INSERT INTO use|
|00000010| 72 20 28 69 64 2c 6e 61 6d 65 29 20 56 41 4c 55 |r (id,name) VALU|
|00000020| 45 53 28 31 31 30 2c 20 27 61 61 61 27 29       |ES(110, 'aaa')  |
+--------+-------------------------------------------------+----------------+
2022-02-14 10:59:06.082 DEBUG 22687 --- [actor-tcp-nio-2] o.s.r2dbc.core.DefaultDatabaseClient     : Executing SQL statement [INSERT INTO user (id,name) VALUES(111, 'aaa')]
2022-02-14 10:59:06.082 TRACE 22687 --- [actor-tcp-nio-2] o.s.r2dbc.core.DefaultDatabaseClient     : Expanded SQL [INSERT INTO user (id,name) VALUES(111, 'aaa')]
2022-02-14 10:59:06.082 DEBUG 22687 --- [actor-tcp-nio-2] dev.miku.r2dbc.mysql.MySqlConnection     : Create a simple statement provided by text query
2022-02-14 10:59:06.082 DEBUG 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : Request: SimpleQueryMessage{sql=REDACTED}
2022-02-14 10:59:06.083 TRACE 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : [id: 0xc62edac7, L:/127.0.0.1:61272 - R:localhost/127.0.0.1:3306] WRITE: 46B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 49 4e 53 45 52 54 20 49 4e 54 4f 20 75 73 65 |.INSERT INTO use|
|00000010| 72 20 28 69 64 2c 6e 61 6d 65 29 20 56 41 4c 55 |r (id,name) VALU|
|00000020| 45 53 28 31 31 31 2c 20 27 61 61 61 27 29       |ES(111, 'aaa')  |
+--------+-------------------------------------------------+----------------+
2022-02-14 10:59:06.083 DEBUG 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.MessageDuplexCodec    : Decode context change to DecodeContext-Command
2022-02-14 10:59:06.083 TRACE 22687 --- [actor-tcp-nio-2] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.example.service.UserServiceImpl.insertRows]
2022-02-14 10:59:06.083 DEBUG 22687 --- [actor-tcp-nio-2] o.s.r.c.R2dbcTransactionManager          : Initiating transaction commit
2022-02-14 10:59:06.083 DEBUG 22687 --- [actor-tcp-nio-2] o.s.r.c.R2dbcTransactionManager          : Committing R2DBC transaction on Connection [PooledConnection[dev.miku.r2dbc.mysql.MySqlConnection@4f99ad97]]
2022-02-14 10:59:06.083 DEBUG 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : Request: SimpleQueryMessage{sql=REDACTED}
2022-02-14 10:59:06.084 TRACE 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : [id: 0xc62edac7, L:/127.0.0.1:61272 - R:localhost/127.0.0.1:3306] WRITE: 4B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 07 00 00 00                                     |....            |
+--------+-------------------------------------------------+----------------+
2022-02-14 10:59:06.084 TRACE 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : [id: 0xc62edac7, L:/127.0.0.1:61272 - R:localhost/127.0.0.1:3306] WRITE: 7B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 43 4f 4d 4d 49 54                            |.COMMIT         |
+--------+-------------------------------------------------+----------------+

2022-02-14 10:59:06.098 TRACE 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : [id: 0xc62edac7, L:/127.0.0.1:61272 - R:localhost/127.0.0.1:3306] WRITE: 17B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 53 45 54 20 61 75 74 6f 63 6f 6d 6d 69 74 3d |.SET autocommit=|
|00000010| 31                                              |1               |

2022-02-14 10:59:06.099 DEBUG 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.MessageDuplexCodec    : Decode context change to DecodeContext-Command
2022-02-14 10:59:06.099 DEBUG 22687 --- [actor-tcp-nio-2] o.s.r.c.R2dbcTransactionManager          : Releasing R2DBC Connection [PooledConnection[dev.miku.r2dbc.mysql.MySqlConnection@4f99ad97]] after transaction
2022-02-14 10:59:06.099 DEBUG 22687 --- [actor-tcp-nio-2] io.r2dbc.pool.PooledConnection           : Releasing connection


若執(zhí)行異常，需要回滾時(shí)，日志如下：
2022-02-14 11:10:31.410 DEBUG 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : Response: ErrorMessage{errorCode=1062, sqlState='23000', errorMessage='Duplicate entry '110' for key 'PRIMARY''}
2022-02-14 11:10:31.419 TRACE 22687 --- [actor-tcp-nio-2] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.example.service.UserServiceImpl.insertRows] after exception: org.springframework.dao.DataIntegrityViolationException: execute; SQL [INSERT INTO user (id,name) VALUES(110, 'aaa')]; Duplicate entry '110' for key 'PRIMARY'; nested exception is io.r2dbc.spi.R2dbcDataIntegrityViolationException: [1062] [23000] Duplicate entry '110' for key 'PRIMARY'
2022-02-14 11:10:31.421 DEBUG 22687 --- [actor-tcp-nio-2] o.s.r.c.R2dbcTransactionManager          : Initiating transaction rollback
2022-02-14 11:10:31.421 DEBUG 22687 --- [actor-tcp-nio-2] o.s.r.c.R2dbcTransactionManager          : Rolling back R2DBC transaction on Connection [PooledConnection[dev.miku.r2dbc.mysql.MySqlConnection@47e30501]]
2022-02-14 11:10:31.421 DEBUG 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : Request: SimpleQueryMessage{sql=REDACTED}
2022-02-14 11:10:31.422 TRACE 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : [id: 0x55800452, L:/127.0.0.1:61274 - R:localhost/127.0.0.1:3306] WRITE: 4B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 09 00 00 00                                     |....            |
+--------+-------------------------------------------------+----------------+
2022-02-14 11:10:31.422 TRACE 22687 --- [actor-tcp-nio-2] d.m.r.mysql.client.ReactorNettyClient    : [id: 0x55800452, L:/127.0.0.1:61274 - R:localhost/127.0.0.1:3306] WRITE: 9B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 52 4f 4c 4c 42 41 43 4b                      |.ROLLBACK       |
+--------+-------------------------------------------------+----------------+

具體源碼查看，在TransactionAspectSupport類(lèi)中。

其中，根據(jù)ReactiveTransactionManager接口類(lèi)型，決定走響應(yīng)式事務(wù)或非響應(yīng)式事務(wù)。下面在編程式事務(wù)中提到的R2dbcTransactionManager就實(shí)現(xiàn)了ReactiveTransactionManager接口。

   protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, final TransactionAspectSupport.InvocationCallback invocation) throws Throwable {
        TransactionAttributeSource tas = this.getTransactionAttributeSource();
        TransactionAttribute txAttr = tas != null ? tas.getTransactionAttribute(method, targetClass) : null;
        TransactionManager tm = this.determineTransactionManager(txAttr);
        if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager) {
            boolean isSuspendingFunction = KotlinDetector.isSuspendingFunction(method);
            boolean hasSuspendingFlowReturnType = isSuspendingFunction && "kotlinx.coroutines.flow.Flow".equals((new MethodParameter(method, -1)).getParameterType().getName());
            if (isSuspendingFunction && !(invocation instanceof TransactionAspectSupport.CoroutinesInvocationCallback)) {
                throw new IllegalStateException("Coroutines invocation not supported: " + method);
            } else {
                TransactionAspectSupport.CoroutinesInvocationCallback corInv = isSuspendingFunction ? (TransactionAspectSupport.CoroutinesInvocationCallback)invocation : null;
                TransactionAspectSupport.ReactiveTransactionSupport txSupport = (TransactionAspectSupport.ReactiveTransactionSupport)this.transactionSupportCache.computeIfAbsent(method, (key) -> {
                    Class<?> reactiveType = isSuspendingFunction ? (hasSuspendingFlowReturnType ? Flux.class : Mono.class) : method.getReturnType();
                    ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(reactiveType);
                    if (adapter == null) {
                        throw new IllegalStateException("Cannot apply reactive transaction to non-reactive return type: " + method.getReturnType());
                    } else {
                        return new TransactionAspectSupport.ReactiveTransactionSupport(adapter);
                    }
                });
                TransactionAspectSupport.InvocationCallback callback = invocation;
                if (corInv != null) {
                    callback = () -> {
                        return CoroutinesUtils.invokeSuspendingFunction(method, corInv.getTarget(), corInv.getArguments());
                    };
                }

                Object result = txSupport.invokeWithinTransaction(method, targetClass, callback, txAttr, (ReactiveTransactionManager)tm);
                if (corInv != null) {
                    Publisher<?> pr = (Publisher)result;
                    return hasSuspendingFlowReturnType ? TransactionAspectSupport.KotlinDelegate.asFlow(pr) : TransactionAspectSupport.KotlinDelegate.awaitSingleOrNull(pr, corInv.getContinuation());
                } else {
                    return result;
                }
            }
        } else {
        
        //略
        }
 }

編程式事務(wù)

與聲明式事務(wù)相比，編程式事務(wù)使用的較少，因?yàn)榕渲孟鄬?duì)繁瑣。但可以幫助理解執(zhí)行邏輯。

 @Resource
    ConnectionFactory connectionFactory;
    
    @Override
    public Mono<Void> insertRows() {
        ReactiveTransactionManager tm=new R2dbcTransactionManager(connectionFactory);
        DatabaseClient db=DatabaseClient.create(connectionFactory);

        TransactionalOperator transactionalOperator=TransactionalOperator.create(tm);
        
        return db.sql("INSERT INTO user (id,name) VALUES(119, 'bbb')")
                .fetch().rowsUpdated()
                .then(db.sql("INSERT INTO user (id,name) VALUES(111, 'aaa')")
                        .then())
                .as(transactionalOperator::transactional);
    }

R2dbcTransactionManager：響應(yīng)式事務(wù)管理器。
DatabaseClient：客戶端使用R2DBC驅(qū)動(dòng)程序提供對(duì)SQL數(shù)據(jù)庫(kù)的訪問(wèn)。
TransactionalOperator：事務(wù)運(yùn)算符將所有上游R2DBC發(fā)布者與事務(wù)上下文相關(guān)聯(lián)。

（3）線程池配置

為了提升數(shù)據(jù)庫(kù)的執(zhí)行效率，減少建立連接的開(kāi)銷(xiāo)，一般數(shù)據(jù)庫(kù)連接都會(huì)有連接池的概念，同樣的r2dbc也有一個(gè)叫做r2dbc-pool的連接池。

<dependency>
 <groupId>io.r2dbc</groupId>
 <artifactId>r2dbc-pool</artifactId>
 <version>${version}</version>
</dependency>

五、遠(yuǎn)程調(diào)用

Spring webflux提供WebClient進(jìn)行遠(yuǎn)程調(diào)用，這里使用webclient做示例，okhttp可以后面作補(bǔ)充。

簡(jiǎn)單服務(wù)

 WebClient webClient= WebClient.builder().build();
 @GetMapping("/remote/webClient")
    public Flux<User> queryByWebclient(){
        log.info("invoke start");
        Flux<User> list=webClient.get().uri(url).retrieve().bodyToFlux(User.class);
        log.info("invoke end");
        return list;
    }

主要執(zhí)行流程：

main thread 打印 invoke start invoke end

創(chuàng)建線程池，進(jìn)行遠(yuǎn)程調(diào)用，初始化pipeline。

連接調(diào)用url，建立channel。

接收響應(yīng)，釋放channel。


2022-02-15 13:58:32.231 DEBUG 27846 --- [ctor-http-nio-5] s.w.r.r.m.a.RequestMappingHandlerMapping : [7a8c56a7-1, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] Mapped to com.example.controller.MyController#queryByWebclient()
2022-02-15 13:58:32.231  INFO 27846 --- [ctor-http-nio-5] com.example.controller.MyController      : invoke start
2022-02-15 13:58:32.232  INFO 27846 --- [ctor-http-nio-5] com.example.controller.MyController      : invoke end
2022-02-15 13:58:32.234 DEBUG 27846 --- [ctor-http-nio-5] o.s.w.r.r.m.a.ResponseBodyResultHandler  : [7a8c56a7-1, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] Using 'application/json' given [text/html, application/xhtml+xml, image/avif, image/webp, image/apng, application/xml;q=0.9, application/signed-exchange;v=b3;q=0.9, */*;q=0.8] and supported [application/json, application/*+json, application/x-ndjson, text/event-stream]
2022-02-15 13:58:32.234 DEBUG 27846 --- [ctor-http-nio-5] o.s.w.r.r.m.a.ResponseBodyResultHandler  : [7a8c56a7-1, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] 0..N [com.example.model.User]
2022-02-15 13:58:32.236 TRACE 27846 --- [ctor-http-nio-5] o.s.w.r.f.client.ExchangeFunctions       : [6c2270b1] HTTP GET http://localhost:8082/web/test?str=webclient, headers={}
2022-02-15 13:58:32.247 DEBUG 27846 --- [ctor-http-nio-5] r.n.resources.PooledConnectionProvider   : [ae492444] Created a new pooled channel, now: 0 active connections, 0 inactive connections and 0 pending acquire requests.
2022-02-15 13:58:32.247 DEBUG 27846 --- [ctor-http-nio-5] reactor.netty.transport.TransportConfig  : [ae492444] Initialized pipeline DefaultChannelPipeline{(reactor.left.httpCodec = io.netty.handler.codec.http.HttpClientCodec), (reactor.left.httpDecompressor = io.netty.handler.codec.http.HttpContentDecompressor), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler)}
2022-02-15 13:58:32.252 DEBUG 27846 --- [ctor-http-nio-5] r.netty.transport.TransportConnector     : [ae492444] Connecting to [localhost/127.0.0.1:8082].
2022-02-15 13:58:32.252 DEBUG 27846 --- [ctor-http-nio-5] r.n.r.DefaultPooledConnectionProvider    : [ae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] Registering 。 release on close event for channel
2022-02-15 13:58:32.252 DEBUG 27846 --- [ctor-http-nio-5] r.n.resources.PooledConnectionProvider   : [ae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] Channel connected, now: 1 active connections, 0 inactive connections and 0 pending acquire requests.
2022-02-15 13:58:32.252 DEBUG 27846 --- [ctor-http-nio-5] r.n.r.DefaultPooledConnectionProvider    : [ae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] onStateChange(PooledConnection{channel=[id: 0xae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082]}, [connected])
2022-02-15 13:58:32.252 DEBUG 27846 --- [ctor-http-nio-5] r.n.r.DefaultPooledConnectionProvider    : [ae492444-1, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] onStateChange(GET{uri=/, connection=PooledConnection{channel=[id: 0xae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082]}}, [configured])
2022-02-15 13:58:32.253 DEBUG 27846 --- [ctor-http-nio-5] r.netty.http.client.HttpClientConnect    : [ae492444-1, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] Handler is being applied: {uri=http://localhost:8082/web/test?str=webclient, method=GET}
2022-02-15 13:58:32.253 DEBUG 27846 --- [ctor-http-nio-5] r.n.r.DefaultPooledConnectionProvider    : [ae492444-1, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] onStateChange(GET{uri=/web/test?str=webclient, connection=PooledConnection{channel=[id: 0xae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082]}}, [request_prepared])
2022-02-15 13:58:32.254 DEBUG 27846 --- [ctor-http-nio-5] r.n.r.DefaultPooledConnectionProvider    : [ae492444-1, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] onStateChange(GET{uri=/web/test?str=webclient, connection=PooledConnection{channel=[id: 0xae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082]}}, [request_sent])
2022-02-15 13:58:33.384 DEBUG 27846 --- [ctor-http-nio-5] r.n.http.client.HttpClientOperations     : [ae492444-1, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] Received response (auto-read:false) : [Date=Tue, 15 Feb 2022 05:58:33 GMT, content-length=0]
2022-02-15 13:58:33.384 DEBUG 27846 --- [ctor-http-nio-5] r.n.r.DefaultPooledConnectionProvider    : [ae492444-1, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] onStateChange(GET{uri=/web/test?str=webclient, connection=PooledConnection{channel=[id: 0xae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082]}}, [response_received])
2022-02-15 13:58:33.384 TRACE 27846 --- [ctor-http-nio-5] o.s.w.r.f.client.ExchangeFunctions       : [6c2270b1] [ae492444-1, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] Response 200 OK, headers={masked}
2022-02-15 13:58:33.385 DEBUG 27846 --- [ctor-http-nio-5] reactor.netty.channel.FluxReceive        : [ae492444-1, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] FluxReceive{pending=0, cancelled=false, inboundDone=false, inboundError=null}: subscribing inbound receiver
2022-02-15 13:58:33.386 DEBUG 27846 --- [ctor-http-nio-5] r.n.http.client.HttpClientOperations     : [ae492444-1, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] Received last HTTP packet
2022-02-15 13:58:33.387 TRACE 27846 --- [ctor-http-nio-5] reactor.netty.channel.ChannelOperations  : [ae492444, L:/127.0.0.1:61048 - 
2022-02-15 13:58:33.432 TRACE 27846 --- [ctor-http-nio-5] o.s.http.codec.json.Jackson2JsonEncoder  : [7a8c56a7-1, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] Encoding [[]]
2022-02-15 13:58:33.433 DEBUG 27846 --- [ctor-http-nio-5] r.n.r.DefaultPooledConnectionProvider    : [ae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] onStateChange(GET{uri=/web/test?str=webclient, connection=PooledConnection{channel=[id: 0xae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082]}}, [response_completed])
2022-02-15 13:58:33.434 DEBUG 27846 --- [ctor-http-nio-5] r.n.r.DefaultPooledConnectionProvider    : [ae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] onStateChange(GET{uri=/web/test?str=webclient, connection=PooledConnection{channel=[id: 0xae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082]}}, [disconnecting])
2022-02-15 13:58:33.434 DEBUG 27846 --- [ctor-http-nio-5] r.n.r.DefaultPooledConnectionProvider    : [ae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] Releasing channel
2022-02-15 13:58:33.434 DEBUG 27846 --- [ctor-http-nio-5] r.n.resources.PooledConnectionProvider   : [ae492444, L:/127.0.0.1:61048 - R:localhost/127.0.0.1:8082] Channel cleaned, now: 0 active connections, 1 inactive connections and 0 pending acquire requests.
2022-02-15 13:58:33.434 TRACE 27846 --- [ctor-http-nio-5] o.s.w.s.adapter.HttpWebHandlerAdapter    : [7a8c56a7-1, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] Completed 200 OK, headers={masked}
2022-02-15 13:58:33.434 TRACE 27846 --- [ctor-http-nio-5] o.s.h.s.r.ReactorHttpHandlerAdapter      : [7a8c56a7-1, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] Handling completed
2022-02-15 13:58:33.434 DEBUG 27846 --- [ctor-http-nio-5] r.n.http.server.HttpServerOperations     : [7a8c56a7-1, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] Last HTTP response frame
2022-02-15 13:58:33.434 DEBUG 27846 --- [ctor-http-nio-5] r.n.http.server.HttpServerOperations     : [7a8c56a7-1, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] Decreasing pending responses, now 0
2022-02-15 13:58:33.434 DEBUG 27846 --- [ctor-http-nio-5] r.n.http.server.HttpServerOperations     : [7a8c56a7-1, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] Last HTTP packet was sent, terminating the channel
2022-02-15 13:58:33.435 TRACE 27846 --- [ctor-http-nio-5] reactor.netty.channel.ChannelOperations  : [7a8c56a7, L:/0:0:0:0:0:0:0:1:20001 - R:/0:0:0:0:0:0:0:1:61046] Disposing ChannelOperation from a channel

首次進(jìn)行遠(yuǎn)程調(diào)用，會(huì)創(chuàng)建一個(gè)http的客戶端線程池，再次發(fā)起遠(yuǎn)程調(diào)用時(shí)，會(huì)直接使用現(xiàn)有線程池。

leasingStrategy：FIFO 先進(jìn)先出。

evictionInterval：實(shí)例銷(xiāo)毀的間隔 0s。

maxConnections：最大連接數(shù)，500。

maxIdleTime: 最大空閑時(shí)間

maxLifeTime: 最大存活時(shí)間

metricsEnabled: 是否啟用指標(biāo)

pendingAcquireMaxCount:

pendingAcquireTimeout:

Creating a new [http] client pool [PoolFactory{evictionInterval=PT0S, leasingStrategy=fifo, maxConnections=500, maxIdleTime=-1, maxLifeTime=-1, metricsEnabled=false, pendingAcquireMaxCount=1000, pendingAcquireTimeout=45000}] for [localhost:8082]

Created a new pooled channel, now: 0 active connections, 0 inactive connections and 0 pending acquire requests.

依賴服務(wù)

如果兩個(gè)遠(yuǎn)程調(diào)用方法之間有彼此依賴關(guān)系，應(yīng)該如何使用？比如rpcB需要使用rpcA返回的結(jié)果值做處理。

示例：

@GetMapping("/remote/webClient")
    public Flux<Integer> queryByWebclient(){
        log.info("invoke start");
        long start =System.nanoTime();
        Flux<Integer> userFlux = userService.getListDependency(0 ).flatMap(k -> userService.getListDependency(k)
        );
        userFlux.subscribe(k-> System.out.println(" controller time"+(System.nanoTime()-start)));
        log.info("invoke end");
        return userFlux;
    }
//遠(yuǎn)程調(diào)用的service
public Flux<Integer> getListDependency(Integer str) {
        Flux<Integer> list=webClient.get().uri(url+str).retrieve().bodyToFlux(Integer.class);
        list.subscribe(k-> System.out.print("service time"+new Date()));
        return list;
 }

六、與Spring MVC的比較

示例

SpringMVC方式

@GetMapping("/remote/webClient")
    public Flux<Integer> queryByWebclient(){
        log.info("invoke start");
        long start =System.nanoTime();
        Flux<Integer> userFlux = userService.getListDependency(0 ).flatMap(k -> userService.getListDependency(k)
        );
        userFlux.subscribe(k-> System.out.println(" controller time"+(System.nanoTime()-start)));
        log.info("invoke end");
        return userFlux;
    }
//遠(yuǎn)程調(diào)用的service
public Flux<Integer> getListDependency(Integer str) {
        Flux<Integer> list=webClient.get().uri(url+str).retrieve().bodyToFlux(Integer.class);
        list.subscribe(k-> System.out.print("service time"+new Date()));
        return list;
 }

日志執(zhí)行結(jié)果：

在這里插入圖片描述

WebFlux的響應(yīng)式處理：

 @GetMapping("/list/reactor")
    public Flux<User> getListReactor(){
        log.info("enter controller");
        Flux<User> listFromDB = userService.getListByReactor();
        log.info("end controller");
        return listFromDB;
    }

@Override
    public  Flux<User> getListByReactor() {
        log.info("enter service");
        List<User> userList = new ArrayList<>();
        Set<Map.Entry<String,User>> entries =userMap.entrySet();
        entries.stream().forEach(entry->userList.add(entry.getValue()));
        Flux<User> userFlux = Flux.fromStream(() -> createStr(userList));
        log.info("end service");
        return userFlux;
    }

  private Stream<User> createStr(List<User> userList) {
        try {
            System.out.println("start sleep");
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
        }
        return userList.stream();
    }

執(zhí)行結(jié)果：

在這里插入圖片描述

小結(jié)：

1、WebFlux是異步處理的，SpringMVC是同步的，但也可以實(shí)現(xiàn)異步非阻塞。

2、Servlet3.1已經(jīng)支持異步非阻塞了，但無(wú)論是回調(diào)或是使用CompletableFuture，代碼實(shí)現(xiàn)起來(lái)會(huì)比較復(fù)雜。WebFlux提供的API會(huì)更加方便。Spring WebFlux在應(yīng)對(duì)高并發(fā)的請(qǐng)求時(shí)，借助于異步IO，能夠以少量而穩(wěn)定的線程處理更高吞吐量的請(qǐng)求。

七、擴(kuò)展

應(yīng)用：

Gateway是基于Webflux實(shí)現(xiàn)的，它通過(guò)擴(kuò)展HandlerMapping與WebHandler來(lái)處理用戶的請(qǐng)求，先通過(guò)Predicate定位到Router然后在經(jīng)過(guò)FilterChain的過(guò)濾處理，最后定位到下層服務(wù)。

以上就是spring webflux響應(yīng)式編程使用詳解的詳細(xì)內(nèi)容，更多關(guān)于spring webflux使用的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

相關(guān)文章

  • ByteArrayOutputStream簡(jiǎn)介和使用_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理

    ByteArrayOutputStream簡(jiǎn)介和使用_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理

    ByteArrayOutputStream 是字節(jié)數(shù)組輸出流。它繼承于OutputStream。這篇文章主要介紹了ByteArrayOutputStream簡(jiǎn)介和使用,需要的朋友可以參考下
    2017-05-05
  • 詳解java中String值為空字符串與null的判斷方法

    詳解java中String值為空字符串與null的判斷方法

    這篇文章主要介紹了詳解java中String值為空字符串與null的判斷方法，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
    2021-04-04
  • Java獲取接口的所有實(shí)現(xiàn)類(lèi)方法總結(jié)示例

    Java獲取接口的所有實(shí)現(xiàn)類(lèi)方法總結(jié)示例

    這篇文章主要給大家介紹了關(guān)于Java獲取接口的所有實(shí)現(xiàn)類(lèi)方法的相關(guān)資料,文中通過(guò)代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下
    2024-06-06
  • Jenkins系統(tǒng)如何進(jìn)行數(shù)據(jù)備份

    Jenkins系統(tǒng)如何進(jìn)行數(shù)據(jù)備份

    隨著我們的長(zhǎng)期使用，Jenkins系統(tǒng)中的內(nèi)容會(huì)越來(lái)越多，特別是一些配置相關(guān)的東西，不能有任何丟失。這個(gè)時(shí)候我們就需要定期備份我們的Jenkins系統(tǒng)，避免一些誤操作不小心刪除了某些重要文件，本文就將介紹下Jenkins系統(tǒng)如何進(jìn)行數(shù)據(jù)備份
    2021-06-06
  • Java常見(jiàn)的3種文件上傳方法和速度對(duì)比

    Java常見(jiàn)的3種文件上傳方法和速度對(duì)比

    這篇文章介紹了Java常見(jiàn)的3種文件上傳方法和速度對(duì)比，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
    2021-12-12
  • java Spring AOP詳解及簡(jiǎn)單實(shí)例

    java Spring AOP詳解及簡(jiǎn)單實(shí)例

    這篇文章主要介紹了java Spring AOP詳解及簡(jiǎn)單實(shí)例的相關(guān)資料,需要的朋友可以參考下
    2017-05-05
  • java?MongoDB實(shí)現(xiàn)列表分頁(yè)查詢的示例代碼

    java?MongoDB實(shí)現(xiàn)列表分頁(yè)查詢的示例代碼

    本文主要介紹了java?MongoDB實(shí)現(xiàn)列表分頁(yè)查詢的示例代碼，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
    2023-07-07
  • SpringBoot異常處理之異常顯示的頁(yè)面問(wèn)題

    SpringBoot異常處理之異常顯示的頁(yè)面問(wèn)題

    這篇文章主要介紹了SpringBoot異常處理異常顯示的頁(yè)面的問(wèn)題，本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下
    2022-09-09
  • springboot2.0?@Slf4j?log?彩色日志配置輸出到文件

    springboot2.0?@Slf4j?log?彩色日志配置輸出到文件

    這篇文章主要介紹了springboot2.0 @Slf4j log日志配置輸出到文件(彩色日志),解決方式是使用了springboot原生自帶的一個(gè)log框架，結(jié)合實(shí)例代碼給大家講解的非常詳細(xì)，需要的朋友可以參考下
    2023-08-08
  • Java中Parser的用法

    Java中Parser的用法

    這篇文章主要介紹了Java?Parser使用指南,本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下
    2023-05-05

最新評(píng)論