SpringBoot 整合WebFlux

女爷i 2024-04-01 13:20 29阅读 0赞

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));
            }
    
    
            @Resource
            JpaEntityServiceImpl 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
    {
        
    
    
        @Resource
        private 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
    {
        
    
        @Resource
        private 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
    {
        
    
        @PersistenceContext
        private 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

@Autowired
        private ServerProperties serverProperties;
    
        @Bean
        public 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.UsersServiceCoroutineStub
    
        override fun filter(exchange: ServerWebExchange, chain: WebFilterChain): Mono<Void> = mono {
        
            val request = exchange.request
    
            if (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_JSON
            response.statusCode = code.httpStatus
    
            return 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