SpringBoot 整合WebFlux

webmvc和webflux作为spring framework的两个重要模块，代表了两个IO模型，阻塞式和非阻塞式的。

webmvc是基于servlet的阻塞式模型（一般称为oio），一个请求到达服务器后会单独分配一个线程去处理请求，如果请求包含IO操作，线程在IO操作结束之前一直处于阻塞等待状态，这样线程在等待IO操作结束的时间就浪费了。

WebFlux
Spring WebFlux 是一个异步非阻塞式的 Web 框架，它能够充分利用多核 CPU 的硬件资源去处理大量的并发请求。

WebFlux 内部使用的是响应式编程（Reactive Programming），以 Reactor 库为基础, 基于异步和事件驱动，可以让我们在不扩充硬件资源的前提下，提升系统的吞吐量和伸缩性。WebFlux 并不能使接口的响应时间缩短，它仅仅能够提升吞吐量和伸缩性。

Spring Reactor
Spring Reactor 是一个反应式库，用于根据反应式流规范在 JVM 上构建非阻塞应用。它是完全非阻塞的，支持反应流背压，并在 Netty，Undertow 和 Servlet 3.1+容器等服务器上运行。

Reactor 项目提供两种类型的发布者：

Flux 是产生 0 到 N 个值的发布者，返回多个元素的操作使用此类型。
Mono 是产生 0 到 1 值的发布者，它用于返回单个元素的操作。

1、添加依赖，版本自行选择

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

2、请求接入
方式1: 注解方式开发

package com.crazymaker.springcloud.reactive.user.info.controller;import com.crazymaker.springcloud.common.result.RestOut;
import com.crazymaker.springcloud.reactive.user.info.dto.User;
import com.crazymaker.springcloud.reactive.user.info.service.impl.JpaEntityServiceImpl;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;import javax.annotation.Resource;/*** Mono 和 Flux 适用于两个场景，即：* Mono：实现发布者，并返回 0 或 1 个元素，即单对象。* Flux：实现发布者，并返回 N 个元素，即 List 列表对象。* 有人会问，这为啥不直接返回对象，比如返回 City/Long/List。* 原因是，直接使用 Flux 和 Mono 是非阻塞写法，相当于回调方式。* 利用函数式可以减少了回调，因此会看不到相关接口。这恰恰是 WebFlux 的好处：集合了非阻塞 + 异步*/
@Slf4j
@Api(value = "用户信息、基础学习DEMO", tags = {"用户信息DEMO"})
@RestController
@RequestMapping("/api/user")
public class UserReactiveController
{@ApiOperation(value = "回显测试", notes = "提示接口使用者注意事项", httpMethod = "GET")@RequestMapping(value = "/hello")@ApiImplicitParams({@ApiImplicitParam(paramType = "query", dataType="string",dataTypeClass = String.class, name = "name",value = "名称", required = true)})public Mono<RestOut<String>> hello(@RequestParam(name = "name") String name){log.info("方法 hello 被调用了");return  Mono.just(RestOut.succeed("hello " + name));}@ResourceJpaEntityServiceImpl jpaEntityService;@PostMapping("/add/v1")@ApiOperation(value = "插入用户" )@ApiImplicitParams({
//                @ApiImplicitParam(paramType = "body", dataType="java.lang.Long", name = "userId", required = false),
//                @ApiImplicitParam(paramType = "body", dataType="用户", name = "dto", required = true)@ApiImplicitParam(paramType = "body",dataTypeClass = User.class, dataType="User", name = "dto",  required = true),})
//    @ApiImplicitParam(paramType = "body", dataType="com.crazymaker.springcloud.reactive.user.info.dto.User",  required = true)public Mono<User> userAdd(@RequestBody User dto){//命令式写法
//        jpaEntityService.delUser(dto);//响应式写法return Mono.create(cityMonoSink -> cityMonoSink.success(jpaEntityService.addUser(dto)));}@PostMapping("/del/v1")@ApiOperation(value = "响应式的删除")@ApiImplicitParams({@ApiImplicitParam(paramType = "body", dataType="User",dataTypeClass = User.class,name = "dto",  required = true),})public Mono<User> userDel(@RequestBody User dto){//命令式写法//        jpaEntityService.delUser(dto);//响应式写法return Mono.create(cityMonoSink -> cityMonoSink.success(jpaEntityService.delUser(dto)));}@PostMapping("/list/v1")@ApiOperation(value = "查询用户")public Flux<User> listAllUser(){log.info("方法 listAllUser 被调用了");//命令式写法 改为响应式 以下语句，需要在流中执行
//        List<User> list = jpaEntityService.selectAllUser();//响应式写法Flux<User> userFlux = Flux.fromIterable(jpaEntityService.selectAllUser());return userFlux;}@PostMapping("/detail/v1")@ApiOperation(value = "响应式的查看")@ApiImplicitParams({@ApiImplicitParam(paramType = "body", dataTypeClass = User.class,dataType="User", name = "dto",  required = true),})public Mono<User> getUser(@RequestBody User dto){log.info("方法 getUser 被调用了");//构造流Mono<User> userMono = Mono.justOrEmpty(jpaEntityService.selectOne(dto.getUserId()));return userMono;}@PostMapping("/detail/v2")@ApiOperation(value = "命令式的查看")@ApiImplicitParams({@ApiImplicitParam(paramType = "body", dataType="User",dataTypeClass = User.class, name = "dto",  required = true),})        public RestOut<User> getUserV2(@RequestBody User dto){log.info("方法 getUserV2 被调用了");User user = jpaEntityService.selectOne(dto.getUserId());return RestOut.success(user);}}

方式2: 配置模式进行WebFlux 接口开发

package com.crazymaker.springcloud.reactive.user.info.config.handler;import com.crazymaker.springcloud.common.exception.BusinessException;
import com.crazymaker.springcloud.reactive.user.info.dto.User;
import com.crazymaker.springcloud.reactive.user.info.service.impl.JpaEntityServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
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;import javax.annotation.Resource;import static org.springframework.http.MediaType.APPLICATION_JSON_UTF8;
import static org.springframework.web.reactive.function.server.ServerResponse.ok;@Slf4j
@Component
public class UserReactiveHandler
{@Resourceprivate JpaEntityServiceImpl jpaEntityService;/*** 得到所有用户** @param request* @return*/public Mono<ServerResponse> getAllUser(ServerRequest request){log.info("方法 getAllUser 被调用了");return ok().contentType(APPLICATION_JSON_UTF8).body(Flux.fromIterable(jpaEntityService.selectAllUser()), User.class);}/*** 创建用户** @param request* @return*/public Mono<ServerResponse> createUser(ServerRequest request){// 2.0.0 是可以工作, 但是2.0.1 下面这个模式是会报异常Mono<User> user = request.bodyToMono(User.class);/**Mono 使用响应式的,时候都是一个流,是一个发布者,任何时候都不能调用发布者的订阅方法也就是不能消费它, 最终的消费还是交给我们的Springboot来对它进行消费,任何时候不能调用它的user.subscribe();不能调用block把异常放在统一的地方来处理*/return user.flatMap(dto ->{// 校验代码需要放在这里if (StringUtils.isBlank(dto.getName())){throw new BusinessException("用户名不能为空");}return ok().contentType(APPLICATION_JSON_UTF8).body(Mono.create(cityMonoSink -> cityMonoSink.success(jpaEntityService.addUser(dto))), User.class);});}/*** 根据id删除用户** @param request* @return*/public Mono<ServerResponse> deleteUserById(ServerRequest request){String id = request.pathVariable("id");// 校验代码需要放在这里if (StringUtils.isBlank(id)){throw new BusinessException("id不能为空");}User dto = new User();dto.setUserId(Long.parseLong(id));return ok().contentType(APPLICATION_JSON_UTF8).body(Mono.create(cityMonoSink -> cityMonoSink.success(jpaEntityService.delUser(dto))), User.class);}}

3、service


package com.crazymaker.springcloud.reactive.user.info.service.impl;import com.crazymaker.springcloud.common.util.BeanUtil;
import com.crazymaker.springcloud.reactive.user.info.dao.impl.JpaUserRepositoryImpl;
import com.crazymaker.springcloud.reactive.user.info.dto.User;
import com.crazymaker.springcloud.reactive.user.info.entity.UserEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;
import java.util.List;@Slf4j
@Service
@Transactional
public class JpaEntityServiceImpl
{@Resourceprivate JpaUserRepositoryImpl userRepository;@Transactional//增加用户public User addUser(User dto){User userEntity = new UserEntity();userEntity.setUserId(dto.getUserId());userEntity.setName(dto.getName());userRepository.insert(userEntity);BeanUtil.copyProperties(userEntity,dto);return dto;}@Transactional//删除用户public User delUser(User dto){userRepository.delete(dto.getUserId());return dto;}//查询全部用户public List<User> selectAllUser(){log.info("方法 selectAllUser 被调用了");return userRepository.selectAll();}//查询一个用户public User selectOne(final Long userId){log.info("方法 selectOne 被调用了");return userRepository.selectOne(userId);}}

4、DAO

package com.crazymaker.springcloud.reactive.user.info.dao.impl;import com.crazymaker.springcloud.reactive.user.info.dto.User;
import org.springframework.stereotype.Repository;import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import javax.persistence.Query;
import javax.transaction.Transactional;
import java.util.List;@Repository
@Transactional
public class JpaUserRepositoryImpl
{@PersistenceContextprivate EntityManager entityManager;public Long insert(final User user){entityManager.persist(user);return user.getUserId();}public void delete(final Long userId){Query query = entityManager.createQuery("DELETE FROM UserEntity o WHERE o.userId = ?1");query.setParameter(1, userId);query.executeUpdate();}@SuppressWarnings("unchecked")public List<User> selectAll(){return (List<User>) entityManager.createQuery("SELECT o FROM UserEntity o").getResultList();}@SuppressWarnings("unchecked")public User selectOne(final Long userId){Query query = entityManager.createQuery("SELECT o FROM UserEntity o WHERE o.userId = ?1");query.setParameter(1, userId);return (User) query.getSingleResult();}
}

5、启动类

@Slf4j
@SpringBootApplication
public class DemoApplication {public static void main(String[] args) {//可以使用以下两种创建SpringApplication的方式SpringApplication application = new SpringApplication(DemoApplication.class);application.run(args);//这种方式，使用SpringApplicationBuilder来创建SpringApplication，builder提供了链式调用API，更加方便，可读性更强//new SpringApplicationBuilder().web(WebApplicationType.REACTIVE).sources(DemoApplication.class).run(args);}
}

6、启动项目
容器已经从默认的 Tomcat 缓存了 webflux 默认的 Netty

2022-04-21 15:15:55.791  INFO 38672 --- [           main] o.s.b.web.embedded.netty.NettyWebServer  : Netty started on port 8080
2022-04-21 15:15:55.798  INFO 38672 --- [           main] .m.w.MingYueSpringbootWebfluxApplication : Started MingYueSpringbootWebfluxApplication in 2.47 seconds (JVM running for 3.95)

7、访问测试
webFlux中dispatchservlet会失效，所以context-path也会无法使用，访问的路径变成了：http://ip:port/getUser，如何让context-path生效，继续使用http://ip:port/contoxt-path/getUser进行访问，有两个解决方法有
方法1:

spring.webflux.base-path=/pageHelper2

方法2:

server.servlet.context-path=/pageHelper

  @Autowiredprivate ServerProperties serverProperties;@Beanpublic WebFilter contextPathWebFilter() {String contextPath = serverProperties.getServlet().getContextPath();return (exchange, chain) -> {ServerHttpRequest request = exchange.getRequest();if (request.getURI().getPath().startsWith(contextPath)) {return chain.filter(exchange.mutate().request(request.mutate().contextPath(contextPath).build()).build());}return chain.filter(exchange);};}

依据上述方式进行配置之后，就可以使用以下地址访问http://localhost:8082/pageHelper/getUser

特别注意：
A. Spring WebFlux并不能使接口的响应时间缩短，它仅是能够提升吞吐量和伸缩性；

B. Spring WebFlux内部使用的是响应式编程，以Reactor库为基础，基于异步和事件驱动，特别适合应用在IO密集型的服务中，如网关；

C. Spring WebFlux并不是Spring MVC的替代方案；

D. Spring WebFlux默认情况下使用Netty作为服务器，不支持MySQL；

E. Spring WebFlux的前端控制器是DispatcherHandler，而Spring MVC是DispatcherServlet；

F. Spring WebFlux支持两种编程风格，一种是Spring MVC的注解形式，另一种就是Java 8 Lambda函数式编程。

G. Reactor类型
　　A. Mono：返回0或者1个元素，即单个对象；
　　B. Flux：返回N个元素，即List列表对象。

H. 使用方式和一般的MVC程序没有什么区别，除了一点,方法需要是suspend方法或返回Mono/Flux

@RestController
class ResourceController(private val resourceService: ResourceService
) {@PostMapping("resources/:push")suspend fun push(@RequestBody pushRequest: PushRequest): PushResponse {val result = resourceService.validateAndSave(pushRequest.resources)return PushResponse(result.map { it.data as Map<String, Any> })}
}

I. Webflux中没有拦截器这个概念，要做类似的工作需要在过滤器中完成，项目中我们用到Token验证，使用方法是注册过滤器

@Component
class AuthFilter(applicationContext: ApplicationContext) : AbstractAuthFilter(applicationContext) {@Value("\${authentication.token.name}")lateinit var tokenName: String// 注意这里使用了协程的grpc stub@GrpcClient("user-service")lateinit var userStub: UsersServiceGrpcKt.UsersServiceCoroutineStuboverride fun filter(exchange: ServerWebExchange, chain: WebFilterChain): Mono<Void> = mono {val request = exchange.requestif (request.needAuth()) {val token = request.headers[tokenName]?.first() val result = userStub.verify(Token.newBuilder().setToken(token).build())... ...}// chain.filter返回的是Mono，需要调用await方法转换为协程chain.filter(exchange).awaitSingleOrNull()}}

J. 全局异常处理
Webflux中可以使用@ControllerAdvice注册全局异常处理器，但它仅Controller中抛出的异常生效，无法顾及到过滤器。对异常，推荐的方式是注册WebExceptionHandler

@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
class ExceptionHandler : ErrorWebExceptionHandler {private val objectMapper = ObjectMapper()// 对协程的支持override fun handle(exchange: ServerWebExchange, ex: Throwable): Mono<Void> {val errResponse = objectMapper.writeValueAsBytes("error message")response.headers.contentType = MediaType.APPLICATION_PROBLEM_JSONresponse.statusCode = code.httpStatusreturn response.writeWith(Mono.just(response.bufferFactory().wrap(errResponse)))}
}

K. 同步DAO的调用
JDBC是同步的，基于它的MyBatis也是同步的，为了不阻塞DIspatcher-Worker线程，需要将其手动调度到其他线程池。当然如下步骤也可以使用AOP实现，这样就不用为每个方法手动调mono方法

// 注入Scheduler
@Autowired
private lateinit var scheduler: Scheduler
// 讲同步代码注册到该scheduler
protected fun <T> mono(block: () -> T): Mono<T> {return Mono.defer { Mono.just(block()) }.subscribeOn(scheduler)
}// 调用方式
fun save(modifications: List<Resource>): Mono<List<Resource>> = mono {modifications.mapNotNull {resourceMapper.save(it)}
}

L. Swagger
Knife4j的增强功能无法在Webflux下使用，且当controller为suspend方法时无法正常读取到返回值，需要打如下补丁。

/***  修复controller方法为suspend方法时，springfox无法获取返回值类型的情况*  因为suspend方法转换为字节码后返回值为null*  #issue: https://github.com/springfox/springfox/issues/3241*/
@Component
@Primary
class CustomRequestHandler(private val resolver: TypeResolver
) : HandlerMethodResolver(resolver) {override fun methodReturnType(handlerMethod: HandlerMethod): ResolvedType {val func = handlerMethod.beanType.kotlin.declaredFunctions.first { it.javaMethod == handlerMethod.method }if (func.returnType == Unit::class.starProjectedType) resolver.resolve(Void.TYPE)return resolver.resolve(func.returnType.javaType)}
}

M. WebApplicationType类型介绍

public enum WebApplicationType {/*** 不启动内嵌的WebServer，不是运行web application*/NONE,/*** 启动内嵌的基于servlet的web server*/SERVLET,/*** 启动内嵌的reactive web server，这个application是一个reactive web application*/REACTIVE;private static final String[] SERVLET_INDICATOR_CLASSES = { "javax.servlet.Servlet","org.springframework.web.context.ConfigurableWebApplicationContext" };private static final String WEBMVC_INDICATOR_CLASS = "org.springframework.web.servlet.DispatcherServlet";private static final String WEBFLUX_INDICATOR_CLASS = "org.springframework.web.reactive.DispatcherHandler";private static final String JERSEY_INDICATOR_CLASS = "org.glassfish.jersey.servlet.ServletContainer";private static final String SERVLET_APPLICATION_CONTEXT_CLASS = "org.springframework.web.context.WebApplicationContext";private static final String REACTIVE_APPLICATION_CONTEXT_CLASS = "org.springframework.boot.web.reactive.context.ReactiveWebApplicationContext";static WebApplicationType deduceFromClasspath() {// 尝试加载org.springframework.web.reactive.DispatcherHandler，如果成功并且加载org.springframework.web.servlet.DispatcherServlet和org.glassfish.jersey.servlet.ServletContainer失败，则这个application是WebApplicationType.REACTIVE类型。if (ClassUtils.isPresent(WEBFLUX_INDICATOR_CLASS, null) && !ClassUtils.isPresent(WEBMVC_INDICATOR_CLASS, null)&& !ClassUtils.isPresent(JERSEY_INDICATOR_CLASS, null)) {return WebApplicationType.REACTIVE;}for (String className : SERVLET_INDICATOR_CLASSES) {// 如果ClassLoader里面同时加载这两个 javax.servlet.Servlet和 org.springframework.web.context.ConfigurableWebApplicationContext成功。则application是WebApplicationType.NONE 类型。if (!ClassUtils.isPresent(className, null)) { return WebApplicationType.NONE;}}// application是 WebApplicationType.SERVLET 类型。return WebApplicationType.SERVLET;}static WebApplicationType deduceFromApplicationContext(Class<?> applicationContextClass) {if (isAssignable(SERVLET_APPLICATION_CONTEXT_CLASS, applicationContextClass)) {return WebApplicationType.SERVLET;}if (isAssignable(REACTIVE_APPLICATION_CONTEXT_CLASS, applicationContextClass)) {return WebApplicationType.REACTIVE;}return WebApplicationType.NONE;}//省略其他代码

参考文档：
https://blog.csdn.net/crazymakercircle/article/details/112977951?spm=1001.2101.3001.6650.1&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7ERate-1-112977951-blog-124322939.pc_relevant_vip_default&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7ERate-1-112977951-blog-124322939.pc_relevant_vip_default&utm_relevant_index=2

https://blog.csdn.net/mo_long00/article/details/125584199

SpringBoot 整合WebFlux相关推荐

SpringBoot整合阿里Druid数据源及Spring-Data-Jpa
SpringBoot整合阿里Druid数据源及Spring-Data-Jpa https://mp.weixin.qq.com/s?__biz=MzU0MDEwMjgwNA==&mid=224 ...
SpringBoot整合Sa-Token
SpringBoot整合Sa-Token 入门介绍技术理论技术实践入门介绍 Sa-Token是一个轻量级Java权限认证框架,主要解决:登录认证.权限认证.Session会话.单点登录.OAut ...
SpringBoot第九篇： springboot整合Redis
这篇文章主要介绍springboot整合redis,至于没有接触过redis的同学可以看下这篇文章:5分钟带你入门Redis. 引入依赖: 在pom文件中添加redis依赖: <dependen ...
es springboot 不设置id_原创 | 一篇解决Springboot 整合 Elasticsearch
ElasticSearch 结合业务的场景,在目前的商品体系需要构建搜索服务,主要是为了提供用户更丰富的检索场景以及高速,实时及性能稳定的搜索服务. ElasticSearch是一个基于Lucene的 ...
springboot整合shiro使用shiro-spring-boot-web-starter
此文章仅仅说明在springboot整合shiro时的一些坑,并不是教程增加依赖  <dependency><groupId> ...
db2 springboot 整合_springboot的yml配置文件通过db2的方式整合mysql的教程
springboot整合MySQL很简单,多数据源就master,slave就行了,但是在整合DB2就需要另起一行,以下是同一个yml文件先配置MySQL,代码如下 spring: datasour ...
九、springboot整合rabbitMQ
springboot整合rabbitMQ 简介 rabbitMQ是部署最广泛的开源消息代理. rabbitMQ轻量级,易于在内部和云中部署. 它支持多种消息传递协议. RabbitMQ可以部署在分布式 ...
八、springboot整合Spring Security
springboot整合Spring Security 简介 Spring Security是一个功能强大且可高度自定义的身份验证和访问控制框架.它是保护基于Spring的应用程序的事实标准. Spr ...
六、springboot整合swagger
六.springboot整合swagger 简介 swagger 提供最强大,最易用的工具,以充分利用OpenAPI规范. 官网 : https://swagger.io/ 准备工作 pom.xml ...

SpringBoot 整合WebFlux

SpringBoot 整合WebFlux相关推荐

最新文章

热门文章