SpringWebFlux介紹
簡介
- SpringWebFlux是Spring5添加的新模塊,用於Web開發,功能和SpringMvc類似的,WebFlux使用當前一種比較流行的響應式編程框架
- 使用傳統的Web框架,比如SpringMvc,這些是基於Servlet容器, WebFlux是一種異步非阻塞的框架,異步非阻塞的框架在Servlet3.1以后才支持,核心是基於Reactor的相關API實現的
什么是異步非阻塞
- 異步和同步
- 非阻塞和阻塞
- 上面都是針對對象不一樣
異步和同步針對調度者,調用者發送請求,如果等待對方回應之后才去做其他事情,就是同步,如果發送請求之后不等着對方回應就去做其他事情就是異步
阻塞和非阻塞針對被調度者,被調度者收到請求后,做完請求任務之后才給出反饋就是阻塞,收到請求之后馬上給出反饋然后去做事情,就是非阻塞
WebFlux特點
- 非阻塞式: 在有限資源下,提高系統吞吐量和伸縮性,以Reactor為基礎實現響應式編程
- 函數式編程: Spring5框架基於Java8, WebFlux使用Java8函數式編程方式實現路由請求
比較SpringMvc
- 第一: 兩個框架都可以使用注解方式,都運行在Tomcat等容器中
- 第二: SpringMvc采用命令式編程, WebFlux采用異步響應式編程
響應式編程(Java實現)
什么是響應式編程
響應式編程是一種面向數據流和變化傳播的編程范式,這意味着可以在編程語言中很方便的表達靜態或動態的數據流,而相關的計算模型會自動將變化的值通過數據流進行傳播,電子表格程序就是響應式編程的一個例子,單元格可以包含字面值或類似"=B1+C1"的公式,而包含公式的單元格的值會依據其他單元格的變化而變化.
Java8及其之前的版本
提供的 "觀察者模式" 兩個類Observer 和 Observable
新建SpringBoot項目
新建一個SpringBoot的項目,我直接建一個模塊算了
ok了
編寫代碼
package com.dance.webflux.reactor8; import lombok.extern.slf4j.Slf4j; import java.util.Observable; import java.util.Observer; @Slf4j public class ObserverDemo extends Observable { public static void main(String[] args) { ObserverDemo observerDemo = new ObserverDemo(); // 添加觀察者 observerDemo.addObserver((o, arg) -> { log.info("o:{},arg:{}",o,arg); System.out.println("發生變化"); }); // 添加觀察者 observerDemo.addObserver((o, arg) -> { log.info("o:{},arg:{}",o,arg); System.out.println("手動被觀察者通知,准備改變"); }); // 發生改變 observerDemo.setChanged(); // 通知觀察者 observerDemo.notifyObservers(); } }
執行結果
23:00:10.650 [main] INFO com.dance.webflux.reactor8.ObserverDemo - o:com.dance.webflux.reactor8.ObserverDemo@504bae78,arg:null 手動被觀察者通知,准備改變 23:00:10.663 [main] INFO com.dance.webflux.reactor8.ObserverDemo - o:com.dance.webflux.reactor8.ObserverDemo@504bae78,arg:null 發生變化
響應式編程(Reactor實現)
簡介
- 響應式編程操作中,Reactor是滿足Reactive規范框架
- Reactor有兩個核心類, Mono和Flux,這兩個類實現接口Publisher,提供豐富的操作符,Flux對象實現發布者,返回N個元素,Mono對象實現發布者,返回1或者0個元素
- Flux和Mono都是數據流的發布者,使用Flux和Mono都可以發出三種數據信號,"元素值","錯誤信號","完成信號",錯誤信號和完成信號都代表終止信號,終止信號用於告訴訂閱者數據流結束了,錯誤信號終止數據流同時把錯誤信息傳遞給訂閱者
代碼演示Flux和Mono
引入依賴
<dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-core</artifactId> <version>3.1.5.RELEASE</version> </dependency>
編寫代碼
package com.dance.webflux.reactor8; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.util.Arrays; import java.util.List; import java.util.stream.Stream; public class TestReactor { public static void main(String[] args) { /* * just 方法直接聲明 */ Flux flux = Flux.just(1, 2, 3); Mono mono = Mono.just(1); // 其他方法 // 數組 Integer[] array = new Integer[]{1,2,3,4}; Flux flux1 = Flux.fromArray(array); // 集合 List array2 = Arrays.asList(1, 2, 3, 4); Flux flux2 = Flux.fromIterable(array2); // Stream Stream stream = array2.stream(); Flux flux3 = Flux.fromStream(stream); // 供給型函數式接口Stream Flux tFlux = Flux.fromStream(() -> Stream.of(1, 2, 3)); } }
三種信號特點
- 錯誤信號和完成信號都是終止信號, 不能共存的
- 如果沒有發送任何元素值,而是直接發送錯誤或者完成信號,表示是空數據流
- 如果沒有錯誤信號,沒有完成信號,表示是無限數據流
真的,去看一下Java8吧,不然真看不懂
訂閱數據流
調用just或者其他方法只是聲明數據流,數據流並沒有發出,只有在進行訂閱之后才會觸發數據流,不訂閱什么都不會發生
// 訂閱數據流 flux.subscribe(x -> System.out.print(x + " ")); System.out.println(); mono.subscribe(System.out::println);
執行結果
1 2 3 1
操作符
對數據進行一道道操作,稱為操作符,比如工廠流水線
第一: map 元素映射為新元素(來自StreamAPI)
第二 flatmap 元素映射為流
把每個元素轉換為流 把轉換之后多個流合並為一個流
SpringWebFlux執行流程和核心API
SpringWebFlux 基於Reactor,默認使用容器是Netty, Netty是高性能NIO框架,異步非阻塞的框架
Netty
BIO
執行過程
SpringWebFlux執行過程和SpringMvc相似
SpringWebFlux 核心控制器 DispatchHandler,實現接口WebHandler
接口WebHandler有一個接口
因為沒有添加依賴所以IDEA中找不到(我找了好長時間)
添加WebFlux依賴
<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-webflux --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency>
FAQ
如果添加依賴失敗可以添加阿里雲倉庫
<repositories> <repository> <id>aliyun</id> <url>https://maven.aliyun.com/repository/public</url> <releases> <enabled>true</enabled> </releases> <snapshots> <enabled>false</enabled> </snapshots> </repository> </repositories> <pluginRepositories> <pluginRepository> <id>aliyun-plugin</id> <url>https://maven.aliyun.com/repository/public</url> <releases> <enabled>true</enabled> </releases> <snapshots> <enabled>false</enabled> </snapshots> </pluginRepository> </pluginRepositories>
WebHandler
// // Source code recreated from a .class file by IntelliJ IDEA // (powered by FernFlower decompiler) // package org.springframework.web.server; import reactor.core.publisher.Mono; public interface WebHandler { Mono handle(ServerWebExchange exchange); }
可以看出返回的就是Mono,一個或零個元素
DispatchHandler實現
@Override public Mono handle(ServerWebExchange exchange) { // http請求 // 判斷請求映射集合是否為空 if (this.handlerMappings == null) { return createNotFoundError(); } // 獲取Request 判斷是否存在前置處理 if (CorsUtils.isPreFlightRequest(exchange.getRequest())) { return handlePreFlight(exchange); } /** * 將映射集合轉為Flux發布(Flux.fromIterable),通過映射集合中的映射獲取匹配, * 然后判斷匹配完成后是否為空,為空返回沒有找到(switchIfEmpty),然后流化執行handler處理器 * (invokeHandler),然后執行返回結果處理(handleResult) 返回一個或零個元素 Mono * / return Flux.fromIterable(this.handlerMappings) .concatMap(mapping -> mapping.getHandler(exchange)) .next() .switchIfEmpty(createNotFoundError()) .flatMap(handler -> invokeHandler(exchange, handler)) .flatMap(result -> handleResult(exchange, result)); }
組件介紹
DispatchHandler: 負責請求的處理
HandlerMapping: 請求映射處理
HandlerAdapter: 請求適配處理
HandlerResultHandler: 響應結果處理
函數式編程接口
SpringWebFlux 實現函數式編程,兩個接口,RouteFunction(路由處理)和HandlerFunction(處理函數)
SpringWebFlux(基於注解編程模型)
- SpringWebFlux實現方式有兩種: 注解編程模型和函數式編程模型
- 使用注解編程模型方式,和之前SpringMvc使用類似的,只需要把相關依賴配置到項目中,SpringBoot自動配置相關運行容器,默認情況下使用Netty服務器
創建SpringBoot項目,引入WebFlux-Starter
前面已經創建和引入了
配置文件修改
application.properties
server.port=8081
創建實體類
package com.dance.webflux.entity; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @Data @AllArgsConstructor @NoArgsConstructor public class User { private String name; private String sex; private Integer age; }
創建Service
新建UserService
package com.dance.webflux.service; import com.dance.webflux.entity.User; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; public interface UserService { /** * 根據ID獲取用戶信息 * @param id id * @return user */ Mono getUserById(int id); /** * 獲取全部用戶信息 * @return 用戶信息 */ Flux getAllUser(); /** * 保存用戶信息 * @param user 用戶信息 * @return void */ Mono saveUserInfo(Mono user); }
實現接口
package com.dance.webflux.service.impl; import com.dance.webflux.entity.User; import com.dance.webflux.service.UserService; import org.springframework.stereotype.Service; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.util.HashMap; import java.util.Map; @Service public class UserServiceImpl implements UserService { /** * 創建 map 集合存儲數據,模擬數據庫 */ private final Map<Integer,User> users = new HashMap<>(); { this.users.put(1,new User("lucy","nan",20)); this.users.put(2,new User("mary","nv",30)); this.users.put(3,new User("jack","nv",50)); } @Override public Mono getUserById(int id) { // 返回一個或者0個元素 return Mono.justOrEmpty(users.get(id)); } @Override public Flux getAllUser() { // 返回多個元素,返回全部的值 return Flux.fromIterable(this.users.values()); } @Override public Mono saveUserInfo(Mono user) { // 處理數據后返回空 return user.doOnNext(x -> { int key = users.size() + 1; users.put(key,x); }).thenEmpty(Mono.empty()); } }
創建Controller
package com.dance.webflux.controller; import com.dance.webflux.entity.User; import com.dance.webflux.service.UserService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.*; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @RestController public class UserController { @Autowired private UserService userService; @GetMapping("/getUserById/{id}") public Mono getUserById(@PathVariable Integer id){ return userService.getUserById(id); } @GetMapping("/getAllUser") public Flux getAllUser(){ return userService.getAllUser(); } @PostMapping("/saveUserInfo") public Mono saveUserInfo(@RequestBody User user){ return userService.saveUserInfo(Mono.just(user)); } }
啟動項目
FAQ,我在啟動的時候報錯了,經過排查后是應為之前為了看類的使用引入的依賴的原因,導致JAR包沖突了
<dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-core</artifactId> <version>3.0.3.RELEASE</version> </dependency>
將這個JAR包注釋掉后,重新啟動就ok了
10:46:27.550 [Thread-1] DEBUG org.springframework.boot.devtools.restart.classloader.RestartClassLoader - Created RestartClassLoader org.springframework.boot.devtools.restart.classloader.RestartClassLoader@7d031891 . ____ _ __ _ _ /\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \ ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \ \\/ ___)| |_)| | | | | || (_| | ) ) ) ) ' |____| .__|_| |_|_| |_\__, | / / / / =========|_|==============|___/=/_/_/_/ :: Spring Boot :: (v2.6.1) 2021-12-13 10:46:27.952 INFO 21120 --- [ restartedMain] com.dance.webflux.WebfluxApplication : Starting WebfluxApplication using Java 1.8.0_162 on ZB-PF2P9QVH with PID 21120 (D:\zhangyugen@JD.com\coding\Spring5\webflux\target\classes started by ext.caiyuanqing in D:\zhangyugen@JD.com\coding\Spring5) 2021-12-13 10:46:27.954 INFO 21120 --- [ restartedMain] com.dance.webflux.WebfluxApplication : No active profile set, falling back to default profiles: default 2021-12-13 10:46:27.996 INFO 21120 --- [ restartedMain] .e.DevToolsPropertyDefaultsPostProcessor : Devtools property defaults active! Set 'spring.devtools.add-properties' to 'false' to disable 2021-12-13 10:46:27.996 INFO 21120 --- [ restartedMain] .e.DevToolsPropertyDefaultsPostProcessor : For additional web related logging consider setting the 'logging.level.web' property to 'DEBUG' 2021-12-13 10:46:28.777 INFO 21120 --- [ restartedMain] o.s.b.d.a.OptionalLiveReloadServer : LiveReload server is running on port 35729 2021-12-13 10:46:29.151 INFO 21120 --- [ restartedMain] o.s.b.web.embedded.netty.NettyWebServer : Netty started on port 8081 2021-12-13 10:46:29.158 INFO 21120 --- [ restartedMain] com.dance.webflux.WebfluxApplication : Started WebfluxApplication in 1.598 seconds (JVM running for 2.507)
測試接口
根據ID獲取用戶信息
http://localhost:8081/getUserById/1
查詢所有用戶
http://localhost:8081/getAllUser
說明
SpringMvc方式實現,同步阻塞的方式, 基於SpringMvc+Servlet+Tomcat
SpringWebFlux方式實現,異步非阻塞的方式,基於SpringWebFlux+Reactor+Netty
SpringWebFlux(基於函數式編程模型)
- 在使用函數式編程模型操作的時候,需要自己初始化服務器
- 基於函數式編程模型的時候,有兩個核心接口,RouterFunction(實現路由功能, 請求轉發給相應的Handler)和HandlerFunction(處理請求生成響應的函數),核心任務定義兩個函數式接口的實現,並且啟動需要的服務器
- SpringWebFlux請求和響應不再是ServletRequest和ServletResponse,而是ServerRequest和ServerResponse
創建Handler(具體實現方法)
package com.dance.webflux.handler; import com.dance.webflux.entity.User; import com.dance.webflux.service.UserService; import org.springframework.http.MediaType; import org.springframework.web.reactive.function.server.ServerRequest; import org.springframework.web.reactive.function.server.ServerResponse; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; public class UserHandler { private final UserService userService; public UserHandler(UserService userService) { this.userService = userService; } /** * 根據ID查詢用戶信息 * @param serverRequest 請求體 * @return 響應體 */ public Mono getUserById(ServerRequest serverRequest){ int id = Integer.parseInt(serverRequest.pathVariable("id")); Mono userById = userService.getUserById(id); return userById.flatMap(user -> ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).bodyValue(user).switchIfEmpty(ServerResponse.notFound().build())); } /** * 查詢所有用戶 * @param serverRequest 請求體 * @return 響應體 */ public Mono getUserAll(ServerRequest serverRequest){ Flux allUser = userService.getAllUser(); return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(allUser,User.class); } /** * 保存用戶信息 * @param serverRequest 請求體 * @return 響應體 */ public Mono saveUserInfo(ServerRequest serverRequest){ Mono userMono = serverRequest.bodyToMono(User.class); return ServerResponse.ok().build(userService.saveUserInfo(userMono)); } }
創建服務器(路由和適配器)
package com.dance.webflux; import com.dance.webflux.handler.UserHandler; import com.dance.webflux.service.impl.UserServiceImpl; import org.springframework.boot.autoconfigure.rsocket.RSocketProperties; import org.springframework.http.MediaType; import org.springframework.http.server.reactive.HttpHandler; import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter; import org.springframework.web.reactive.function.server.RouterFunction; import org.springframework.web.reactive.function.server.RouterFunctions; import org.springframework.web.reactive.function.server.ServerResponse; import reactor.netty.DisposableServer; import reactor.netty.http.server.HttpServer; import java.io.IOException; import static org.springframework.web.reactive.function.server.RequestPredicates.*; import static org.springframework.web.reactive.function.server.RouterFunctions.toHttpHandler; public class FunctionServer { /** * 創建路由 * @return 路由函數 */ public RouterFunction routingFunction(){ // 創建UserService UserServiceImpl userService = new UserServiceImpl(); // 創建處理器 UserHandler userHandler = new UserHandler(userService); // 設置路由 return RouterFunctions .route(GET("/user/{id}").and(accept(MediaType.APPLICATION_JSON)), userHandler::getUserById) .andRoute(GET("/users").and(accept(MediaType.APPLICATION_JSON)), userHandler::getUserAll) .andRoute(POST("/saveUserInfo"), userHandler::saveUserInfo); } /** * 創建服務器完成適配 */ public void createReactorServer(){ // 路由和Handler適配 RouterFunction serverResponseRouterFunction = routingFunction(); // 轉換為HttpHandler HttpHandler httpHandler = toHttpHandler(serverResponseRouterFunction); // 轉換為適配器 ReactorHttpHandlerAdapter reactorHttpHandlerAdapter = new ReactorHttpHandlerAdapter(httpHandler); // 創建HttpServer 設置端口 HttpServer httpServer = HttpServer.create().port(8081); // 指定處理適配器並綁定 DisposableServer disposableServer = httpServer.handle(reactorHttpHandlerAdapter).bindNow(); } /** * 啟動 * @param args 參數 */ public static void main(String[] args) throws IOException { FunctionServer server = new FunctionServer(); server.createReactorServer(); System.out.println("enter to exit"); System.in.read(); } }
啟動項目
啟動main方法
16:06:37.931 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework 16:06:38.773 [main] DEBUG org.springframework.web.server.adapter.HttpWebHandlerAdapter - enableLoggingRequestDetails='false': form data and headers will be masked to prevent unsafe logging of potentially sensitive data 16:06:38.854 [main] DEBUG io.netty.util.internal.logging.InternalLoggerFactory - Using SLF4J as the default logging framework 16:06:38.877 [main] DEBUG io.netty.util.internal.PlatformDependent0 - -Dio.netty.noUnsafe: false 16:06:38.878 [main] DEBUG io.netty.util.internal.PlatformDependent0 - Java version: 8 16:06:38.883 [main] DEBUG io.netty.util.internal.PlatformDependent0 - sun.misc.Unsafe.theUnsafe: available 16:06:38.885 [main] DEBUG io.netty.util.internal.PlatformDependent0 - sun.misc.Unsafe.copyMemory: available 16:06:38.887 [main] DEBUG io.netty.util.internal.PlatformDependent0 - java.nio.Buffer.address: available 16:06:38.888 [main] DEBUG io.netty.util.internal.PlatformDependent0 - direct buffer constructor: available 16:06:38.891 [main] DEBUG io.netty.util.internal.PlatformDependent0 - java.nio.Bits.unaligned: available, true 16:06:38.894 [main] DEBUG io.netty.util.internal.PlatformDependent0 - jdk.internal.misc.Unsafe.allocateUninitializedArray(int): unavailable prior to Java9 16:06:38.894 [main] DEBUG io.netty.util.internal.PlatformDependent0 - java.nio.DirectByteBuffer.(long, int): available 16:06:38.895 [main] DEBUG io.netty.util.internal.PlatformDependent - sun.misc.Unsafe: available 16:06:38.896 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.tmpdir: C:\Users\EXT~1.CAI\AppData\Local\Temp (java.io.tmpdir) 16:06:38.896 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.bitMode: 64 (sun.arch.data.model) 16:06:38.897 [main] DEBUG io.netty.util.internal.PlatformDependent - Platform: Windows 16:06:38.900 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.maxDirectMemory: 3760193536 bytes 16:06:38.900 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.uninitializedArrayAllocationThreshold: -1 16:06:38.902 [main] DEBUG io.netty.util.internal.CleanerJava6 - java.nio.ByteBuffer.cleaner(): available 16:06:38.902 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.noPreferDirect: false 16:06:39.003 [main] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetection.level: simple 16:06:39.003 [main] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetection.targetRecords: 4 16:06:39.115 [main] DEBUG reactor.netty.tcp.TcpResources - [http] resources will use the default LoopResources: DefaultLoopResources {prefix=reactor-http, daemon=true, selectCount=8, workerCount=8} 16:06:39.115 [main] DEBUG reactor.netty.tcp.TcpResources - [http] resources will use the default ConnectionProvider: reactor.netty.resources.DefaultPooledConnectionProvider@26275bef 16:06:39.118 [main] DEBUG reactor.netty.resources.DefaultLoopIOUring - Default io_uring support : false 16:06:39.744 [main] DEBUG reactor.netty.resources.DefaultLoopEpoll - Default Epoll support : false 16:06:39.746 [main] DEBUG reactor.netty.resources.DefaultLoopKQueue - Default KQueue support : false 16:06:39.768 [main] DEBUG io.netty.channel.MultithreadEventLoopGroup - -Dio.netty.eventLoopThreads: 16 16:06:39.819 [main] DEBUG io.netty.util.internal.InternalThreadLocalMap - -Dio.netty.threadLocalMap.stringBuilder.initialSize: 1024 16:06:39.819 [main] DEBUG io.netty.util.internal.InternalThreadLocalMap - -Dio.netty.threadLocalMap.stringBuilder.maxSize: 4096 16:06:39.836 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.noKeySetOptimization: false 16:06:39.836 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.selectorAutoRebuildThreshold: 512 16:06:39.869 [main] DEBUG io.netty.util.internal.PlatformDependent - org.jctools-core.MpscChunkedArrayQueue: available 16:06:39.961 [main] DEBUG io.netty.channel.DefaultChannelId - -Dio.netty.processId: 18728 (auto-detected) 16:06:39.965 [main] DEBUG io.netty.util.NetUtil - -Djava.net.preferIPv4Stack: false 16:06:39.965 [main] DEBUG io.netty.util.NetUtil - -Djava.net.preferIPv6Addresses: false 16:06:40.496 [main] DEBUG io.netty.util.NetUtilInitializations - Loopback interface: lo (Software Loopback Interface 1, 127.0.0.1) 16:06:40.498 [main] DEBUG io.netty.util.NetUtil - Failed to get SOMAXCONN from sysctl and file \proc\sys\net\core\somaxconn. Default: 200 16:06:40.971 [main] DEBUG io.netty.channel.DefaultChannelId - -Dio.netty.machineId: 00:50:56:ff:fe:c0:00:08 (auto-detected) 16:06:41.012 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.numHeapArenas: 16 16:06:41.012 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.numDirectArenas: 16 16:06:41.012 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.pageSize: 8192 16:06:41.012 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.maxOrder: 11 16:06:41.012 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.chunkSize: 16777216 16:06:41.013 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.smallCacheSize: 256 16:06:41.013 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.normalCacheSize: 64 16:06:41.013 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.maxCachedBufferCapacity: 32768 16:06:41.013 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.cacheTrimInterval: 8192 16:06:41.013 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.cacheTrimIntervalMillis: 0 16:06:41.013 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.useCacheForAllThreads: true 16:06:41.013 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.maxCachedByteBuffersPerChunk: 1023 16:06:41.028 [main] DEBUG io.netty.buffer.ByteBufUtil - -Dio.netty.allocator.type: pooled 16:06:41.028 [main] DEBUG io.netty.buffer.ByteBufUtil - -Dio.netty.threadLocalDirectBufferSize: 0 16:06:41.028 [main] DEBUG io.netty.buffer.ByteBufUtil - -Dio.netty.maxThreadLocalCharBufferSize: 16384 16:06:41.216 [reactor-http-nio-1] DEBUG reactor.netty.transport.ServerTransport - [40eb46d7, L:/0:0:0:0:0:0:0:0:62011] Bound new server enter to exit
測試
根據ID獲取用戶信息
http://localhost:8081/user/1
獲取全部用戶信息
http://localhost:8081/users
SpringWebFlux(WebClient調用)
編寫WebClient
package com.dance.webflux; import com.dance.webflux.entity.User; import org.springframework.http.MediaType; import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Flux; public class FunctionClient { public static void main(String[] args) { //調用服務器地址 WebClient webClient = WebClient.create("http://127.0.0.1:8081"); //根據 id 查詢 String id = "1"; User userresult = webClient.get().uri("/user/{id}", id).accept(MediaType.APPLICATION_JSON).retrieve().bodyToMono(User.class).block(); System.out.println(userresult); //查詢所有 Flux results = webClient.get().uri("/users").accept(MediaType.APPLICATION_JSON).retrieve().bodyToFlux(User.class); results.map(User::getName).buffer().doOnNext(System.out::println).blockFirst(); } }
執行結果
16:19:25.382 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework 16:19:25.440 [main] DEBUG io.netty.util.internal.logging.InternalLoggerFactory - Using SLF4J as the default logging framework 16:19:25.446 [main] DEBUG io.netty.util.internal.PlatformDependent0 - -Dio.netty.noUnsafe: false 16:19:25.447 [main] DEBUG io.netty.util.internal.PlatformDependent0 - Java version: 8 16:19:25.449 [main] DEBUG io.netty.util.internal.PlatformDependent0 - sun.misc.Unsafe.theUnsafe: available 16:19:25.450 [main] DEBUG io.netty.util.internal.PlatformDependent0 - sun.misc.Unsafe.copyMemory: available 16:19:25.451 [main] DEBUG io.netty.util.internal.PlatformDependent0 - java.nio.Buffer.address: available 16:19:25.452 [main] DEBUG io.netty.util.internal.PlatformDependent0 - direct buffer constructor: available 16:19:25.453 [main] DEBUG io.netty.util.internal.PlatformDependent0 - java.nio.Bits.unaligned: available, true 16:19:25.453 [main] DEBUG io.netty.util.internal.PlatformDependent0 - jdk.internal.misc.Unsafe.allocateUninitializedArray(int): unavailable prior to Java9 16:19:25.454 [main] DEBUG io.netty.util.internal.PlatformDependent0 - java.nio.DirectByteBuffer.(long, int): available 16:19:25.454 [main] DEBUG io.netty.util.internal.PlatformDependent - sun.misc.Unsafe: available 16:19:25.454 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.tmpdir: C:\Users\EXT~1.CAI\AppData\Local\Temp (java.io.tmpdir) 16:19:25.455 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.bitMode: 64 (sun.arch.data.model) 16:19:25.455 [main] DEBUG io.netty.util.internal.PlatformDependent - Platform: Windows 16:19:25.457 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.maxDirectMemory: 3760193536 bytes 16:19:25.457 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.uninitializedArrayAllocationThreshold: -1 16:19:25.459 [main] DEBUG io.netty.util.internal.CleanerJava6 - java.nio.ByteBuffer.cleaner(): available 16:19:25.459 [main] DEBUG io.netty.util.internal.PlatformDependent - -Dio.netty.noPreferDirect: false 16:19:25.517 [main] DEBUG reactor.netty.tcp.TcpResources - [http] resources will use the default LoopResources: DefaultLoopResources {prefix=reactor-http, daemon=true, selectCount=8, workerCount=8} 16:19:25.517 [main] DEBUG reactor.netty.tcp.TcpResources - [http] resources will use the default ConnectionProvider: reactor.netty.resources.DefaultPooledConnectionProvider@13fee20c 16:19:25.635 [main] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetection.level: simple 16:19:25.636 [main] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetection.targetRecords: 4 16:19:26.686 [main] DEBUG io.netty.util.NetUtil - -Djava.net.preferIPv4Stack: false 16:19:26.687 [main] DEBUG io.netty.util.NetUtil - -Djava.net.preferIPv6Addresses: false 16:19:27.099 [main] DEBUG io.netty.util.NetUtilInitializations - Loopback interface: lo (Software Loopback Interface 1, 127.0.0.1) 16:19:27.100 [main] DEBUG io.netty.util.NetUtil - Failed to get SOMAXCONN from sysctl and file \proc\sys\net\core\somaxconn. Default: 200 16:19:27.138 [main] DEBUG org.springframework.web.reactive.function.client.ExchangeFunctions - [32502377] HTTP GET http://127.0.0.1:8081/user/1 16:19:27.158 [main] DEBUG reactor.netty.resources.DefaultLoopIOUring - Default io_uring support : false 16:19:27.682 [main] DEBUG reactor.netty.resources.DefaultLoopEpoll - Default Epoll support : false 16:19:27.683 [main] DEBUG reactor.netty.resources.DefaultLoopKQueue - Default KQueue support : false 16:19:27.690 [main] DEBUG io.netty.channel.MultithreadEventLoopGroup - -Dio.netty.eventLoopThreads: 16 16:19:27.711 [main] DEBUG io.netty.util.internal.InternalThreadLocalMap - -Dio.netty.threadLocalMap.stringBuilder.initialSize: 1024 16:19:27.711 [main] DEBUG io.netty.util.internal.InternalThreadLocalMap - -Dio.netty.threadLocalMap.stringBuilder.maxSize: 4096 16:19:27.720 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.noKeySetOptimization: false 16:19:27.720 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.selectorAutoRebuildThreshold: 512 16:19:27.732 [main] DEBUG io.netty.util.internal.PlatformDependent - org.jctools-core.MpscChunkedArrayQueue: available 16:19:28.183 [main] DEBUG io.netty.resolver.dns.DefaultDnsServerAddressStreamProvider - Default DNS servers: [/10.2.255.200:53, /172.31.36.36:53, /172.31.37.37:53] (sun.net.dns.ResolverConfiguration) 16:19:28.193 [main] DEBUG reactor.netty.resources.PooledConnectionProvider - Creating a new [http] client pool [PoolFactory{evictionInterval=PT0S, leasingStrategy=fifo, maxConnections=500, maxIdleTime=-1, maxLifeTime=-1, metricsEnabled=false, pendingAcquireMaxCount=1000, pendingAcquireTimeout=45000}] for [/127.0.0.1:8081] 16:19:28.260 [main] DEBUG io.netty.channel.DefaultChannelId - -Dio.netty.processId: 25696 (auto-detected) 16:19:28.611 [main] DEBUG io.netty.channel.DefaultChannelId - -Dio.netty.machineId: 00:50:56:ff:fe:c0:00:08 (auto-detected) 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.numHeapArenas: 16 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.numDirectArenas: 16 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.pageSize: 8192 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.maxOrder: 11 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.chunkSize: 16777216 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.smallCacheSize: 256 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.normalCacheSize: 64 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.maxCachedBufferCapacity: 32768 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.cacheTrimInterval: 8192 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.cacheTrimIntervalMillis: 0 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.useCacheForAllThreads: true 16:19:28.640 [main] DEBUG io.netty.buffer.PooledByteBufAllocator - -Dio.netty.allocator.maxCachedByteBuffersPerChunk: 1023 16:19:28.651 [main] DEBUG io.netty.buffer.ByteBufUtil - -Dio.netty.allocator.type: pooled 16:19:28.652 [main] DEBUG io.netty.buffer.ByteBufUtil - -Dio.netty.threadLocalDirectBufferSize: 0 16:19:28.652 [main] DEBUG io.netty.buffer.ByteBufUtil - -Dio.netty.maxThreadLocalCharBufferSize: 16384 16:19:28.681 [reactor-http-nio-2] DEBUG reactor.netty.resources.PooledConnectionProvider - [da237854] Created a new pooled channel, now: 0 active connections, 0 inactive connections and 0 pending acquire requests. 16:19:28.753 [reactor-http-nio-2] DEBUG io.netty.buffer.AbstractByteBuf - -Dio.netty.buffer.checkAccessible: true 16:19:28.753 [reactor-http-nio-2] DEBUG io.netty.buffer.AbstractByteBuf - -Dio.netty.buffer.checkBounds: true 16:19:28.754 [reactor-http-nio-2] DEBUG io.netty.util.ResourceLeakDetectorFactory - Loaded default ResourceLeakDetector: io.netty.util.ResourceLeakDetector@74ee324f 16:19:28.761 [reactor-http-nio-2] DEBUG reactor.netty.transport.TransportConfig - [da237854] 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)} 16:19:28.786 [reactor-http-nio-2] DEBUG reactor.netty.transport.TransportConnector - [da237854] Connecting to [/127.0.0.1:8081]. 16:19:28.792 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Registering pool release on close event for channel 16:19:28.792 [reactor-http-nio-2] DEBUG reactor.netty.resources.PooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Channel connected, now: 1 active connections, 0 inactive connections and 0 pending acquire requests. 16:19:28.793 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}, [connected]) 16:19:28.806 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [configured]) 16:19:28.808 [reactor-http-nio-2] DEBUG reactor.netty.http.client.HttpClientConnect - [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Handler is being applied: {uri=http://127.0.0.1:8081/user/1, method=GET} 16:19:28.810 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/user/1, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [request_prepared]) 16:19:28.834 [reactor-http-nio-2] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.maxCapacityPerThread: 4096 16:19:28.834 [reactor-http-nio-2] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.maxSharedCapacityFactor: 2 16:19:28.834 [reactor-http-nio-2] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.linkCapacity: 16 16:19:28.834 [reactor-http-nio-2] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.ratio: 8 16:19:28.834 [reactor-http-nio-2] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.delayedQueue.ratio: 8 16:19:28.852 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/user/1, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [request_sent]) 16:19:28.860 [reactor-http-nio-2] DEBUG io.netty.handler.codec.compression.Brotli - brotli4j not in the classpath; Brotli support will be unavailable. 16:19:28.863 [reactor-http-nio-2] DEBUG reactor.netty.http.client.HttpClientOperations - [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Received response (auto-read:false) : [Content-Type=application/json, content-length=36] 16:19:28.863 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/user/1, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [response_received]) 16:19:28.870 [reactor-http-nio-2] DEBUG org.springframework.web.reactive.function.client.ExchangeFunctions - [32502377] [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Response 200 OK 16:19:28.981 [reactor-http-nio-2] DEBUG reactor.netty.channel.FluxReceive - [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] FluxReceive{pending=0, cancelled=false, inboundDone=false, inboundError=null}: subscribing inbound receiver 16:19:28.989 [reactor-http-nio-2] DEBUG reactor.netty.http.client.HttpClientOperations - [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Received last HTTP packet 16:19:29.026 [reactor-http-nio-2] DEBUG org.springframework.http.codec.json.Jackson2JsonDecoder - [32502377] [da237854-1, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Decoded [User(name=lucy, sex=nan, age=20)] User(name=lucy, sex=nan, age=20) 16:19:29.026 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/user/1, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [response_completed]) 16:19:29.026 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/user/1, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [disconnecting]) 16:19:29.026 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Releasing channel 16:19:29.030 [reactor-http-nio-2] DEBUG reactor.netty.resources.PooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Channel cleaned, now: 0 active connections, 1 inactive connections and 0 pending acquire requests. 16:19:29.030 [main] DEBUG org.springframework.web.reactive.function.client.ExchangeFunctions - [58ce9668] HTTP GET http://127.0.0.1:8081/users 16:19:29.032 [reactor-http-nio-2] DEBUG reactor.netty.resources.PooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Channel acquired, now: 1 active connections, 0 inactive connections and 0 pending acquire requests. 16:19:29.032 [reactor-http-nio-2] DEBUG reactor.netty.http.client.HttpClientConnect - [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Handler is being applied: {uri=http://127.0.0.1:8081/users, method=GET} 16:19:29.032 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/users, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [request_prepared]) 16:19:29.033 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/users, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [request_sent]) 16:19:29.037 [reactor-http-nio-2] DEBUG reactor.netty.http.client.HttpClientOperations - [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Received response (auto-read:false) : [transfer-encoding=chunked, Content-Type=application/json] 16:19:29.037 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/users, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [response_received]) 16:19:29.038 [reactor-http-nio-2] DEBUG org.springframework.web.reactive.function.client.ExchangeFunctions - [58ce9668] [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Response 200 OK 16:19:29.058 [reactor-http-nio-2] DEBUG reactor.netty.channel.FluxReceive - [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] FluxReceive{pending=0, cancelled=false, inboundDone=false, inboundError=null}: subscribing inbound receiver 16:19:29.064 [reactor-http-nio-2] DEBUG org.springframework.http.codec.json.Jackson2JsonDecoder - [58ce9668] [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Decoded [User(name=lucy, sex=nan, age=20)] 16:19:29.064 [reactor-http-nio-2] DEBUG org.springframework.http.codec.json.Jackson2JsonDecoder - [58ce9668] [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Decoded [User(name=mary, sex=nv, age=30)] 16:19:29.065 [reactor-http-nio-2] DEBUG org.springframework.http.codec.json.Jackson2JsonDecoder - [58ce9668] [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Decoded [User(name=jack, sex=nv, age=50)] 16:19:29.065 [reactor-http-nio-2] DEBUG reactor.netty.http.client.HttpClientOperations - [da237854-2, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Received last HTTP packet [lucy, mary, jack] 16:19:29.066 [reactor-http-nio-2] DEBUG org.springframework.web.reactive.function.client.ExchangeFunctions - [58ce9668] Cancel signal (to close connection) 16:19:29.066 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/users, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [response_completed]) 16:19:29.066 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] onStateChange(GET{uri=/users, connection=PooledConnection{channel=[id: 0xda237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081]}}, [disconnecting]) 16:19:29.066 [reactor-http-nio-2] DEBUG reactor.netty.resources.DefaultPooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Releasing channel 16:19:29.066 [reactor-http-nio-2] DEBUG reactor.netty.resources.PooledConnectionProvider - [da237854, L:/127.0.0.1:62998 - R:/127.0.0.1:8081] Channel cleaned, now: 0 active connections, 1 inactive connections and 0 pending acquire requests.
里面有很完整的調用鏈信息,完結撒花花
若有收獲,就點個贊吧
作者:彼岸舞
時間:2021\12\13
內容關於:Java
本文屬於作者原創,未經允許,禁止轉發