[Java知识库] SpringBoot 整合WebFlux

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> Java知识库 -> SpringBoot 整合WebFlux -> 正文阅读

[Java知识库]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));
        }


        @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