在Web应用程序中使用RxJava Observables无法解决性能提升

我正在执行一些测试来评估使用基于Observables的反应API是否真的有优势，而不是阻止传统的API。

整个例子可以在Githug上找到

令人惊讶的是，结果表明，输出的结果是：

• 最好的 ：返回包含阻塞操作的Callable / DeferredResult REST服务。

• 同样糟糕 ：阻止REST服务。

• 最糟糕的 ：返回DeferredResult的REST服务，其结果由RxJava Observable设置。

这是我的Spring WebApp：

应用 ：

 @SpringBootApplication public class SpringNioRestApplication { @Bean public ThreadPoolTaskExecutor executor(){ ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10); executor.setMaxPoolSize(20); return executor; } public static void main(String[] args) { SpringApplication.run(SpringNioRestApplication.class, args); } } 

SyncController ：

 @RestController("SyncRestController") @Api(value="", description="Synchronous data controller") public class SyncRestController { @Autowired private DataService dataService; @RequestMapping(value="/sync/data", method=RequestMethod.GET, produces="application/json") @ApiOperation(value = "Gets data", notes="Gets data synchronously") @ApiResponses(value={@ApiResponse(code=200, message="OK")}) public List getData(){ return dataService.loadData(); } } 

AsyncController ：具有原始Callable和Observable端点

 @RestController @Api(value="", description="Synchronous data controller") public class AsyncRestController { @Autowired private DataService dataService; private Scheduler scheduler; @Autowired private TaskExecutor executor; @PostConstruct protected void initializeScheduler(){ scheduler = Schedulers.from(executor); } @RequestMapping(value="/async/data", method=RequestMethod.GET, produces="application/json") @ApiOperation(value = "Gets data", notes="Gets data asynchronously") @ApiResponses(value={@ApiResponse(code=200, message="OK")}) public Callable<List> getData(){ return ( () -> {return dataService.loadData();} ); } @RequestMapping(value="/observable/data", method=RequestMethod.GET, produces="application/json") @ApiOperation(value = "Gets data through Observable", notes="Gets data asynchronously through Observable") @ApiResponses(value={@ApiResponse(code=200, message="OK")}) public DeferredResult<List> getDataObservable(){ DeferredResult<List> dr = new DeferredResult<List>(); Observable<List> dataObservable = dataService.loadDataObservable(); dataObservable.subscribeOn(scheduler).subscribe( dr::setResult, dr::setErrorResult); return dr; } } 

DataServiceImpl

 @Service public class DataServiceImpl implements DataService{ @Override public List loadData() { return generateData(); } @Override public Observable<List> loadDataObservable() { return Observable.create( s -> { List dataList = generateData(); s.onNext(dataList); s.onCompleted(); }); } private List generateData(){ List dataList = new ArrayList(); for (int i = 0; i < 20; i++) { Data data = new Data("key"+i, "value"+i); dataList.add(data); } //Processing time simulation try { Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } return dataList; } } 

我设置了Thread.sleep(500)延迟以增加服务响应时间。

负载测试的结果是：

与Callable的异步 ：700 rps，没有错误

 >>loadtest -c 15 -t 60 --rps 700 http://localhost:8080/async/data ... Requests: 0, requests per second: 0, mean latency: 0 ms Requests: 2839, requests per second: 568, mean latency: 500 ms Requests: 6337, requests per second: 700, mean latency: 500 ms Requests: 9836, requests per second: 700, mean latency: 500 ms ... Completed requests: 41337 Total errors: 0 Total time: 60.002348360999996 s Requests per second: 689 Total time: 60.002348360999996 s 

阻止 ：大约404 rps，但会产生错误

 >>loadtest -c 15 -t 60 --rps 700 http://localhost:8080/sync/data ... Requests: 7683, requests per second: 400, mean latency: 7420 ms Requests: 9683, requests per second: 400, mean latency: 9570 ms Requests: 11680, requests per second: 399, mean latency: 11720 ms Requests: 13699, requests per second: 404, mean latency: 13760 ms ... Percentage of the requests served within a certain time 50% 8868 ms 90% 22434 ms 95% 24103 ms 99% 25351 ms 100% 26055 ms (longest request) 100% 26055 ms (longest request) -1: 7559 errors Requests: 31193, requests per second: 689, mean latency: 14350 ms Errors: 1534, accumulated errors: 7559, 24.2% of total requests 

与Observable的异步 ： 不超过20 rps，并且会更快地获得错误

 >>loadtest -c 15 -t 60 --rps 700 http://localhost:8080/observable/data Requests: 0, requests per second: 0, mean latency: 0 ms Requests: 90, requests per second: 18, mean latency: 2250 ms Requests: 187, requests per second: 20, mean latency: 6770 ms Requests: 265, requests per second: 16, mean latency: 11870 ms Requests: 2872, requests per second: 521, mean latency: 1560 ms Errors: 2518, accumulated errors: 2518, 87.7% of total requests Requests: 6373, requests per second: 700, mean latency: 1590 ms Errors: 3401, accumulated errors: 5919, 92.9% of total requests 

Observable使用10的corePoolSize执行，但将其增加到50并没有改善任何东西。

可能是什么解释？

更新 ：根据akarnokd的建议，我做了以下更改。 在服务中从Object.create移动到Object.fromCallable并重用控制器中的Scheduler，但我仍然得到相同的结果。

Answer 1

问题是由某个时刻的编程错误引起的。 实际上，问题中的例子非常有效。

一个警告，以防止其他人遇到问题：小心使用Observable.just(func) ，func实际上是在Observable创建时调用的。 所以放在那里的任何Thread.sleep都会阻塞调用线程

 @Override public Observable> loadDataObservable() { return Observable.just(generateData()).delay(500, TimeUnit.MILLISECONDS); } private List generateData(){ List dataList = new ArrayList(); for (int i = 0; i < 20; i++) { Data data = new Data("key"+i, "value"+i); dataList.add(data); } return dataList; } 

我在RxJava Google小组开始讨论，他们帮助我解决了这个问题。