spring webflux基本使用案例

一时失言乱红尘 2022-12-01 15:48 297阅读 0赞

### 文章目录 ###

*  使用案例
 *  其他

# 使用案例 #

引入依赖

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

controller层应用方式

@RequestMapping("/getAll")
        public Mono<Response> getAll() { 
            return Mono.justOrEmpty(Response.success("ok"));
        }
    //或者如下（webflux在某些地方兼容springmvc）
    @RequestMapping("/getAll")
        public Response getAll() { 
            return Response.success("ok");
        }

对于某些文件上传需要这样按下面写，（因为webflux没有MultipartFile，MultipartFile是阻塞式的，webflux是非阻塞式的，二者必然不兼容）

@PostMapping("/base/file/add")
        public Mono<ServerResponse> uploadDefault(ServerRequest serverRequest) { 
            return serverRequest
                    .body(BodyExtractors.toMultipartData())
                    .log()
                    .flatMap(parts -> { 
                        return Mono.just((FilePart) 
    ///*文件上传表单字段名*/
    // String UPLOAD_FILE_FORM_FIELD_NAME = "file"; 
                       parts.toSingleValueMap().get(UPLOAD_FILE_FORM_FIELD_NAME));
                    })
                    .flatMap(filePart -> { 
                      // filePart是 FilePart的对象，它装载着二进制文件
                        return fileService.upload(参数1，2，3);
                    });
        }

然后再程序启动时就构造的bean中加入：

@Bean
        public RouterFunction<ServerResponse> routes(FileController fileController) { 
            return   nest(
                    path("/base/file"),
                    route(POST("/add/{type}"), fileController::upload)
                    .andRoute(POST("/add"), fileController::uploadDefault)
                    .andRoute(POST("/initMultipartUpload/{type}"), fileController::initMultipartUpload)
                    .andRoute(POST("/completeMultipartUpload/{type}"), fileController::completeMultipartUpload)
                    .andRoute(POST("/downloadPart"), fileController::downloadPart)
            );
        }

其请求之前的全局拦截器编写如下：

@Slf4j
    @Component
    public class RequestInfoInjectionFilter implements WebFilter { 
    
        @Autowired
        private FileConfigEntity defaultFileConfigEntity;
    
        @Autowired
        private FileConfigRepository fileConfigRepository;
    
        @Value("${auth.header-name}")
        private String authHeaderName;
    
        @Value("${application.header-name}")
        private String appCodeHeader;
    
    
        @Autowired
        private DataBufferWriter dataBufferWriter;
    
        private final AntPathMatcher antPathMatcher = new AntPathMatcher();
    
    
        @SneakyThrows
        @Override
        public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) { 
            String requestUrl = exchange.getRequest().getURI().toString();
            String reqPath = exchange.getRequest().getPath().toString();
            //解决线上生成日志过大问题
            if (reqPath.equals("/")||reqPath.equals("/favicon.ico")){ 
               return exchange.getResponse().writeWith(Mono.justOrEmpty(null));
            }
            //只对非健康的检测的请求做拦截处理
            if (!antPathMatcher.match("**/actuator/health", requestUrl)) { 
                log.info("requestPath:{}",reqPath);
                HttpHeaders httpHeaders = exchange.getRequest().getHeaders();
                String userId = Optional.ofNullable(httpHeaders
                        .getFirst(authHeaderName)).orElse("");
                RequestUtil.setCurrentUser(User.builder().userId(userId).build());
                String accessFlag = Optional.ofNullable(httpHeaders
                        .getFirst("access-flag")).orElse("1");
                RequestUtil.setCurrentAccessFlag(Integer.parseInt(accessFlag));
                log.warn("inject auth info: current user:{}.",
                        RequestUtil.getCurrentUser());
                String curAppCode = Optional.ofNullable(httpHeaders.getFirst(appCodeHeader))
                        .orElse("default");
                RequestUtil.setCurrentAppCode(curAppCode);
                log.warn("inject auth info: current appCode:{}.",
                        RequestUtil.getCurrentAppCode());
                Long requestBodyEstimateSize = httpHeaders.getContentLength();
                RequestUtil.setCurrentRequestBodySize(requestBodyEstimateSize);
                log.warn("ingect request body size info: estimate value{}.",
                        requestBodyEstimateSize);
                if (!RequestUtil.getCurrentAppCode().equals("default")){ 
                    defaultFileConfigEntity=fileConfigRepository.findByAppCode(curAppCode);
                    if (defaultFileConfigEntity==null) throw new OutOfBusinessException("error app-code:"+curAppCode);
                }
                RequestUtil.setCurrenFileConfig(defaultFileConfigEntity);
                Integer allowAnonymousDownload = defaultFileConfigEntity.getAnonDown();
                Integer allowAnonymousUpload = defaultFileConfigEntity.getAnonUpload();
                if (antPathMatcher.match(DOWNLOAD_URL_PATTERN, requestUrl)
                        && !(allowAnonymousDownload==1) && StringUtils.isEmpty(userId)) { 
                    log.warn("capture un-auth download request{} from{}.", requestUrl,
                            exchange.getRequest().getRemoteAddress());
                    return dataBufferWriter.write(exchange.getResponse(), AUTH_FAILED_RESPONSE);
                }
    
                if (antPathMatcher.match(UPLOAD_URL_PATTERN, requestUrl)
                        && !(allowAnonymousUpload==1)
                        && StringUtils.isEmpty(userId)) { 
                    log.warn("capture un-auth upload request,{}.", requestUrl);
                    return dataBufferWriter.write(exchange.getResponse(), AUTH_FAILED_RESPONSE);
                }
                return chain.filter(exchange).doFinally(signalType -> { 
                    RequestUtil.clearCurrentUser();
                    log.warn("clear auth info.");
                    RequestUtil.clearCurrentRequestBodySize();
                    log.warn("clear request body size info.");
                    RequestUtil.clearCurrenFileConfig();
                    log.warn("clear current fileConfig info.");
                });
            }
            return chain.filter(exchange).doFinally(signalType -> { });
        }
    }

# 其他 #

Spring WebFlux 多次读取 DataBuffer 类型的请求内容  
方法一：基于内存缓存 —— 构造新的 DataBuffer 对象

@Override
        public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) { 
            final ServerHttpRequest request = exchange.getRequest(); 
            long contentLength = request.getHeaders().getContentLength();
    
            if (contentLength <= 0) { 
                return chain.filter(exchange);
            }
    
            return DataBufferUtils.join(request.getBody()).map(dataBuffer -> { 
                exchange.getAttributes().put("cachedRequestBody", dataBuffer);
    
                ServerHttpRequest decorator = new ServerHttpRequestDecorator(request) { 
                    @Override
                    public Flux<DataBuffer> getBody() { 
                        return Mono.<DataBuffer>fromSupplier(() -> { 
                            if (exchange.getAttributeOrDefault("cachedRequestBody", null) == null) { 
                                // probably == downstream closed
                                return null;
                            }
    
                            // reset position
                            dataBuffer.readPosition(0);
    
                            // deal with Netty
                            NettyDataBuffer pdb = (NettyDataBuffer) dataBuffer;
                            return pdb.factory().wrap(pdb.getNativeBuffer().retainedSlice());
                        }).flux();
                    }
                };
    
                // TODO 消费 dataBuffer，例如计算 dataBuffer 的哈希值并验证
                // ...
    
                return decorator
    
            })
            .switchIfEmpty(Mono.just(request))
            .flatMap(req -> chain.filter(exchange.mutate().request(req).build()))
            .doFinally(s -> { 
                DataBuffer dataBuffer = exchange.getAttributeOrDefault("cachedRequestBody", null);
                if (dataBuffer != null) { 
                    DataBufferUtils.release(dataBuffer);
                }
            });
        }

方法二：基于内存缓存 —— byte\[\]

@Override
        public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) { 
            final ServerHttpRequest request = exchange.getRequest(); 
            long contentLength = request.getHeaders().getContentLength();
    
            if (contentLength <= 0) { 
                return chain.filter(exchange);
            }
    
            final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    
            return Mono.create(sink -> { 
                 DataBufferUtils.write(request.getBody(), outputStream).subscribe(DataBufferUtils::release, sink::error, sink::success);
            })
            .then(Mono.just(request))
            .flatMap(req -> { 
                log.debug("缓存大小：{}", outputStream.size());
                final ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(req) { 
                    @Override
                    public Flux<DataBuffer> getBody() { 
                        return DataBufferUtils.read(new ByteArrayResource(outputStream.toByteArray()), exchange.getResponse().bufferFactory(), 1024 * 8);
                    }
                };
    
                // TODO 对缓存的 ByteArrayOutputStream 进行处理，例如计算 ByteArrayOutputStream 中 byte[] 的哈希值并验证
                // ...
    
                return chain.filter(exchange.mutate().request(decorator).build());
            });
        }

方法三：基于文件缓存 —— Path

@Override
        public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) { 
            final ServerHttpRequest request = exchange.getRequest(); 
            long contentLength = request.getHeaders().getContentLength();
    
            if (contentLength <= 0) { 
                return chain.filter(exchange);
            }
    
            try { 
                final Path tempFile = Files.createTempFile("HttpRequest", ".bin");
    
                return DataBufferUtils.write(request.getBody(), tempFile)
                    .then(Mono.just(request))
                    .flatMap(req -> { 
                        final ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(req) { 
                            @Override
                            public Flux<DataBuffer> getBody() { 
                                return DataBufferUtils.read(tempFile, exchange.getResponse().bufferFactory(), 1024 * 8, StandardOpenOption.READ);
                            }
                        };
    
                        // TODO 对缓存的 tempFile 进行处理，例如计算 tempFile 的哈希值并验证
                        // ...
    
                        return chain.filter(exchange.mutate().request(decorator).build());
                    })
                    .doFinally(s -> { 
                        try { 
                            Files.deleteIfExists(tempFile);
                        } catch (IOException e) { 
                            throw new IllegalStateException(e);
                        }
                    });
    
            } catch (IOException e) { 
                throw new IllegalStateException(e);
            }
        }

注：

> 在请求 body 比较大的情况的测试中，发现调用 DataBufferUtils\#join() 方法（方法一）会占用较大的内存，并且请求完毕时可能不会立刻释放，在下一次 GC 时可释放。调用 DataBufferUtils\#write() 方法直接写到 OutputStream （方法二）或者临时文件（方法三）时，则不会占用过多内存。

webflux实战增删查改=====视频讲解的是非常好的。

实例文章：写的很好  
https://segmentfault.com/q/1010000018526503  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDYzMTQx_size_16_color_FFFFFF_t_70_pic_center]  
接受json数据还是按照mvc那一套就好了  
返回json可以返回对象。  
但是对于这种mono的处理方式还有待研究

restfor风格的传参  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDYzMTQx_size_16_color_FFFFFF_t_70_pic_center 1]  
数据量很大，建议使用json参数

Mono file.flatMap(s -> \{  
Flux content = s.content();  
// return content;  
// 因为它的类型是flux，然而 Mono file.flatMap()的返回类型是mono，  
// 所以会报如下错误： no instance(s) of type variable(s) R exist so that flux conforms to mono<? extends R>  
// 不存在类型变量R的实例，以至于flux不符合mono<？extends R> 就是当前你返回的值的类型和该函数要求的值的返回类型不兼容  
return null;  
\});

AtomicInteger a= new AtomicInteger();
          Mono<FilePart> file.flatMap(s -> { 
                Flux<DataBuffer> content = s.content();
                content.flatMap(aa->{ 
                    InputStream inputStream = aa.asInputStream();
                    System.out.println("hahhaha"); 
                    a.set(10);
                      return null;
                });
                return null;
            });
            System.out.println(a.get()+"==============");  //打印结果为0

> System.out.println(“hahhaha”);  
> a.set(10);

*  这两条语句虽然执行了，但是不会产生任何作用：即不会改变a的值，也不会输出“haha”,因为他们是流里面的中间操作，除非你返回结果，流会将计算结果存入返回值。否则将会丢弃一切计算。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDYzMTQx_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/39976a1784bc473b89262d6560eeacd1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDYzMTQx_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/c75ae2cb8e0c4000af83c90d4e3f52f1.png