Spring WebFlux运用中的思考与对比

Stella981
• 阅读 1133

系列目录：

  1. Spring WebFlux运用中的思考与对比
  2. CompletableFuture与Spring的Sleuth结合工具类
  3. CommpetableFuture使用anyOf过程中的一些优化思考
  4. 结合CompletableFuture与Spring的Sleuth结合工具类与allOf以及anyOf

本文基于Spring Cloud Finchley SR4

本文通过几个问题，解析下Spring WebFlux用法最佳实践，并与另一框架Vertx作对比

1. 是否一定要用默认的Web容器，用自己的Web容器是否可以，同时是否可以有web和webflux

是可以的，这样的依赖是可行的（容器用tomcat和undertow或者其他都可以，这里使用undertow）：

2. 怎样实现真正的异步背压的Reactor模型呢？

这个问题，除此运用像WebFlux和Vertx的框架的人，都会对这个有误解。认为仅仅简单的把webFlux的依赖添加进来，之后接口返回Mono就实现了异步背压的Reactor模型。实际上并不是这样的。 我们来举几个例子，分步骤深入了解下。 首先为了测试方便，我们将web容器的处理http请求线程池的大小改成唯一一个，对于Tomcat，配置：

server.thread.max-thread=1

对于UnderTow（我们这里用的是underTow）：

# 设置IO线程数, 它主要执行非阻塞的任务,它们会负责多个连接, 默认设置每个CPU核心一个线程
server.undertow.io-threads=1
# 阻塞任务线程池, 当执行类似servlet请求阻塞IO操作, undertow会从这个线程池中取得线程
# 它的值设置取决于系统线程执行任务的阻塞系数，默认值是IO线程数*8
server.undertow.worker-threads=1

之后，配置Log4J2日志格式为：

<Property name="springAppName">test</Property>
    <Property name="LOG_ROOT">log</Property>
    <Property name="LOG_DATEFORMAT_PATTERN">yyyy-MM-dd HH:mm:ss.SSS</Property>
    <Property name="LOG_EXCEPTION_CONVERSION_WORD">%xwEx</Property>
    <Property name="LOG_LEVEL_PATTERN">%5p</Property>
    <Property name="logFormat">
        %d{${LOG_DATEFORMAT_PATTERN}} ${LOG_LEVEL_PATTERN} [${springAppName},%X{X-B3-TraceId},%X{X-B3-SpanId}] [${sys:PID}] [%t][%C:%L]: %m%n${sys:LOG_EXCEPTION_CONVERSION_WORD}
    </Property>

这样的格式可以使我们看到线程号，还有sleuth的traceId和spanId（我们的项目依赖了sleuth）。 首先编写测试代码，看看直接简单调用并just是否实现了异步背压：

@Log4j2
@RestController
public class TestController {
    @Autowired
    private TestService testService;
    
    @RequestMapping("/test")
    public Mono<String> test() {
        log.info("test started");
        return Mono.just(testService.simulateIOTest());
    }

    @Service
    public static class TestService {

        public String simulateIOTest() {
            try {
                //simulate io
                log.info("simulate start");
                TimeUnit.SECONDS.sleep(5);
                log.info("simulate end");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            return "hello";
        }
    }
}

并发调用接口，查看日志，发现：

2019-11-12 09:05:41.595  INFO [test,26bf995af305ad34,26bf995af305ad34] [26208] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController:26]:test started
2019-11-12 09:05:41.596  INFO [test,26bf995af305ad34,26bf995af305ad34] [26208] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController$TestService:38]:simulate start
2019-11-12 09:05:46.598  INFO [test,26bf995af305ad34,26bf995af305ad34] [26208] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController$TestService:40]:simulate end
2019-11-12 09:05:46.635  INFO [test,620bd553b1e55dcd,620bd553b1e55dcd] [26208] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController:26]:test started
2019-11-12 09:05:46.635  INFO [test,620bd553b1e55dcd,620bd553b1e55dcd] [26208] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController$TestService:38]:simulate start
2019-11-12 09:05:51.636  INFO [test,620bd553b1e55dcd,620bd553b1e55dcd] [26208] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController$TestService:40]:simulate end
2019-11-12 09:05:51.643  INFO [test,bc17d60861ba1a2a,bc17d60861ba1a2a] [26208] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController:26]:test started
2019-11-12 09:05:51.643  INFO [test,bc17d60861ba1a2a,bc17d60861ba1a2a] [26208] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController$TestService:38]:simulate start
2019-11-12 09:05:56.644  INFO [test,bc17d60861ba1a2a,bc17d60861ba1a2a] [26208] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController$TestService:40]:simulate end

可以明显看出，请求是串行处理的，因为只有一个线程，并且这个线程还在等待请求处理完。这就不符合Reactor模型，处理http请求的线程XNIO-2 task-1应该不等待请求处理完而直接处理下一个请求才对。 把Mono.just(testService.simulateIOTest())替换成Mono.fromCallable(() -> testService.simulateIOTest())等等类似的是一样的效果，这里必须自己用其他的线程池，去处理实际请求，处理结束的时候，将结果填写到最外层的Mono里面。这样的话，考虑到代码整洁性不采用纯回调写法，要求每一个调用方法返回的都是Future类型的。这里我们返回CompletableFuture。

@Log4j2
@RestController
public class TestController {
    @Autowired
    private TestService testService;

    @RequestMapping("/test")
    public Mono<String> test() {
        log.info("test started");
        return Mono.create(stringMonoSink -> testService.simulateIOTest().thenApply(s -> {
            log.info("apply");
            //填写成功结果
            stringMonoSink.success(s);
            return s;
        }));
    }

    @Service
    public static class TestService {

        public CompletableFuture<String> simulateIOTest() {
            return CompletableFuture.supplyAsync(() -> {
                try {
                    //simulate io
                    log.info("simulate start");
                    TimeUnit.SECONDS.sleep(5);
                    log.info("simulate end");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "hello";
            });
        }
    }
}

结果是：

2019-11-12 09:18:03.457  INFO [test,8d6eddc9cc80612f,8d6eddc9cc80612f] [22892] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController:26]:test started
2019-11-12 09:18:03.458  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-3][com.hopegaming.syringe.api.frontend.order.TestController$TestService:44]:simulate start
2019-11-12 09:18:04.155  INFO [test,c654462e159fd43e,c654462e159fd43e] [22892] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController:26]:test started
2019-11-12 09:18:04.156  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-5][com.hopegaming.syringe.api.frontend.order.TestController$TestService:44]:simulate start
2019-11-12 09:18:04.962  INFO [test,8366a95d002ca25a,8366a95d002ca25a] [22892] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController:26]:test started
2019-11-12 09:18:04.963  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-7][com.hopegaming.syringe.api.frontend.order.TestController$TestService:44]:simulate start
2019-11-12 09:18:05.756  INFO [test,5f851d9e2ef49f14,5f851d9e2ef49f14] [22892] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController:26]:test started
2019-11-12 09:18:05.757  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-9][com.hopegaming.syringe.api.frontend.order.TestController$TestService:44]:simulate start
2019-11-12 09:18:08.459  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-3][com.hopegaming.syringe.api.frontend.order.TestController$TestService:46]:simulate end
2019-11-12 09:18:08.459  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-3][com.hopegaming.syringe.api.frontend.order.TestController:28]:apply
2019-11-12 09:18:09.156  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-5][com.hopegaming.syringe.api.frontend.order.TestController$TestService:46]:simulate end
2019-11-12 09:18:09.156  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-5][com.hopegaming.syringe.api.frontend.order.TestController:28]:apply
2019-11-12 09:18:09.964  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-7][com.hopegaming.syringe.api.frontend.order.TestController$TestService:46]:simulate end
2019-11-12 09:18:09.964  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-7][com.hopegaming.syringe.api.frontend.order.TestController:28]:apply
2019-11-12 09:18:10.757  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-9][com.hopegaming.syringe.api.frontend.order.TestController$TestService:46]:simulate end
2019-11-12 09:18:10.757  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-9][com.hopegaming.syringe.api.frontend.order.TestController:28]:apply

这样，才真正实现了Reactor模型。

3. CompletableFuture线程池管理还有日志追踪

CompletableFuture可以指定线程池，亦可以不指定。如果像上面不指定的话，那么使用的线程池就是Java8之后会默认启动一个大小为CPU核数减一的CommonForkJoinPool去执行。需要指定的话，基本上每个方法都可以额外传入一个线程池作为参数。

最佳实践是，只要涉及到IO的，就交给不同的线程池去做，不同种类的IO的线程池不同。例如，用于数据库IO的线程池，用于RPC的线程池，用于缓存访问的线程池等等。

这里还有一个问题存在，就是异步调用，导致spanId和traceId丢失了，例如上面的例子：

2019-11-12 09:18:03.457  INFO [test,8d6eddc9cc80612f,8d6eddc9cc80612f] [22892] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController:26]:test started
2019-11-12 09:18:03.458  INFO [test,,] [22892] [ForkJoinPool.commonPool-worker-3][com.hopegaming.syringe.api.frontend.order.TestController$TestService:44]:simulate start

8d6eddc9cc80612f这个丢失了，导致微服务调用链日志追踪变得不可行，所以，这里我们对于异步的代码，也需要在异步调用前强制设置下spanId和traceId。

综上之后，修改的代码是：

@Log4j2
@RestController
public class TestController {
    @Autowired
    private TestService testService;

    @RequestMapping("/test")
    public Mono<String> test() {
        log.info("test started");
        return Mono.fromFuture(testService.simulateIOTest());
    }

    @Service
    public static class TestService {
        @Autowired
        private Tracer tracer;
        ThreadFactory build = (new ThreadFactoryBuilder()).setNameFormat("test_service_executor-%d").build();
        private ExecutorService executorService = new ThreadPoolExecutor(50, 50, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(131072), build, new ThreadPoolExecutor.AbortPolicy());

        public CompletableFuture<String> simulateIOTest() {
            Span span = tracer.currentSpan();
            return CompletableFuture.supplyAsync(() -> {
                try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) {
                    //simulate io
                    log.info("simulate start");
                    TimeUnit.SECONDS.sleep(5);
                    log.info("simulate end");
                    return "hello";
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }, executorService);
        }
    }
}

结果是：

2019-11-12 09:44:30.953  INFO [test,bc1ba4169e037577,bc1ba4169e037577] [2796] [XNIO-2 task-1][com.hopegaming.syringe.api.frontend.order.TestController:28]:test started
2019-11-12 09:44:30.991  INFO [test,bc1ba4169e037577,bc1ba4169e037577] [2796] [test_service_executor-0][com.hopegaming.syringe.api.frontend.order.TestController$TestService:44]:simulate start
2019-11-12 09:44:35.991  INFO [test,bc1ba4169e037577,bc1ba4169e037577] [2796] [test_service_executor-0][com.hopegaming.syringe.api.frontend.order.TestController$TestService:46]:simulate end

3. 与Vertx对比，有哪些异同？

实际上，从设计上看，基本思路是一样的。对于任意一个IO操作，如果有原生的异步客户端（返回是一个Future），则运用Future封装交给其他线程池处理，不影响http请求线程接受其他请求。

主要区别在于：

  1. WebFlux框架并没有涉及到线程池，Vertx连异步线程池也封装成为Vertx的WorkerExecutor类。
  2. WebFlux异步Future使用的还是Java原生的，Vetx框架自己封装了Future。
  3. WebFlux与Spring在结合方面更完善，但是Spring生态里面并没有提供原生的NIO客户端，例如实现了MySQL协议栈的NIO mysql客户端，这个Vertx是有的，但是是否健壮还有待考证。这些进一步限制了WebFlux的性能。
  4. Vertx是一个跨语言的框架
线程池testspring框架completablefutureinfo前端前端
点赞
收藏
评论区
推荐文章
blmius blmius
•
3年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时，报错：原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等，这样可以更容易地在不同的环境中使用MySQL。全局s
4069
0
0
helloworld_34035044 helloworld_34035044
•
2年前
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时，如何在新增行里自动增加id呢？能新增整数排序id吗？目前可以在新增行里自动增加id，但只能用uuid函数增加UUID编码，不能新增整数排序id。uuid函数说明：获取一个UUID，可以在填报表中用来创建数据ID语法：uuid()或uuid(sep)参数说明：sep布尔值，生成的uuid中是否包含分隔符''，缺省为
1111
0
0
待兔 待兔
•
3个月前
手写Java HashMap源码
HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22
1043
0
0
Jacquelyn38 Jacquelyn38
•
3年前
2020年前端实用代码段，为你的工作保驾护航
有空的时候，自己总结了几个代码段，在开发中也经常使用，谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
2361
0
0
Stella981 Stella981
•
3年前
CommpetableFuture使用anyOf过程中的一些优化思考
系列目录：1.SpringWebFlux运用中的思考与对比(https://www.oschina.net/action/GoToLink?urlhttps%3A%2F%2Fblog.csdn.net%2Fzhxdick%2Farticle%2Fdetails%2F103035149)2.CompletableFuture与Sprin
831
0
0
Stella981 Stella981
•
3年前
KVM调整cpu和内存
一.修改kvm虚拟机的配置1、virsheditcentos7找到“memory”和“vcpu”标签，将<namecentos7</name<uuid2220a6d1a36a4fbb8523e078b3dfe795</uuid
1324
0
0
Wesley13 Wesley13
•
3年前
mysql设置时区
mysql设置时区mysql\_query("SETtime\_zone'8:00'")ordie('时区设置失败，请联系管理员！');中国在东8区所以加8方法二：selectcount(user\_id)asdevice,CONVERT\_TZ(FROM\_UNIXTIME(reg\_time),'08:00','0
1285
0
0
Stella981 Stella981
•
3年前
CompletableFuture与Spring的Sleuth结合工具类
系列目录：1.SpringWebFlux运用中的思考与对比(https://www.oschina.net/action/GoToLink?urlhttps%3A%2F%2Fblog.csdn.net%2Fzhxdick%2Farticle%2Fdetails%2F103035149)2.CompletableFuture与Sprin
589
0
0
Wesley13 Wesley13
•
3年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
1485
0
0
Python进阶者 Python进阶者
•
9个月前
Excel中这日期老是出来00：00：00，怎么用Pandas把这个去除
大家好，我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下：这日期老是出来00：00：00，怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下：pd.toexcel之前把这
482
0
0
Stella981
Stella981
Lv1
接天莲叶无穷碧，映日荷花别样红。
文章
1.7w
粉丝
42
获赞
28
热门文章
OpenVPN下载、安装、配置及使用详解
GitHub神器，一个可以白嫖全网无损音乐的神器
OpenWrt 路由器过滤广告的N种方法
Python—执行系统命令的四种方法（os.system、os.popen、commands、subprocess）
SS端加密以及obfs混淆