Không thể giải thích được hiệu suất sử dụng RxJava Observables trong các ứng dụng Web

Question

Tôi đang thực hiện một số thử nghiệm để đánh giá xem có lợi thế thực nào trong việc sử dụng API phản ứng dựa trên Observables hay không.

Toàn bộ ví dụ là available on Githug

Đáng ngạc nhiên kết quả cho thấy các kết quả thoughput là:

• tốt nhất: Dịch vụ REST của mà trả về một Callable/DeferredResult rằng kết thúc tốt đẹp các hoạt động ngăn chặn.

• Không phải là xấu: Chặn dịch vụ REST.

• Điều tồi tệ nhất: Dịch vụ REST của mà trả về một DeferredResult mà kết quả được thiết lập bởi một RxJava Quan sát.

Đây là mùa xuân WebApp tôi:

Application:

@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<Data> getData(){ 
     return dataService.loadData(); 
    } 
} 

AsyncController: Với cả hai nguyên Callable và điểm cuối Quan sát

@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<Data>> 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<Data>> getDataObservable(){ 
     DeferredResult<List<Data>> dr = new DeferredResult<List<Data>>(); 
     Observable<List<Data>> dataObservable = dataService.loadDataObservable(); 
     dataObservable.subscribeOn(scheduler).subscribe(dr::setResult, dr::setErrorResult); 
     return dr; 
    } 
} 

DataServiceImpl

@Service 
public class DataServiceImpl implements DataService{ 

    @Override 
    public List<Data> loadData() { 
     return generateData(); 
    } 

    @Override 
    public Observable<List<Data>> loadDataObservable() { 
     return Observable.create(s -> { 
      List<Data> dataList = generateData(); 
      s.onNext(dataList); 
      s.onCompleted(); 
     }); 
    } 

    private List<Data> generateData(){ 
     List<Data> dataList = new ArrayList<Data>(); 
     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; 
    } 
} 

tôi đã thiết lập một sự chậm trễ Thread.sleep(500) để tăng thời gian đáp ứng dịch vụ.

Có kết quả từ các bài kiểm tra tải trọng là:

Async với Callable: 700 rps, không có lỗi

>>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 

Chặn: khoảng 404 rps nhưng tạo ra lỗi

>>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 

Không đồng bộ với Quan sát: không quá 20 rps, và được lỗi sớm

>>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 

Các thực thi Quan sát với một corePoolSize 10, nhưng tăng nó đến 50 không được cải thiện gì cả.

Điều gì có thể là giải thích?

CẬP NHẬT: Theo đề xuất của akarnokd tôi đã thực hiện các thay đổi sau. Chuyển từ Object.create sang Object.fromCallable trong dịch vụ và sử dụng lại Scheduler trong controller, nhưng tôi vẫn nhận được kết quả tương tự.

Answer 1

Sự cố xảy ra do lỗi lập trình tại một số thời điểm. Trên thực tế, ví dụ trong câu hỏi hoạt động hoàn hảo.

Một cảnh báo để ngăn người khác gặp sự cố: hãy cẩn thận khi sử dụng Observable.just(func), func thực sự được gọi là tạo cảnh quan. Vì vậy, bất kỳ Thread.sleep nào được đặt ở đó sẽ chặn chuỗi cuộc gọi

@Override 
public Observable<List<Data>> loadDataObservable() { 
    return Observable.just(generateData()).delay(500, TimeUnit.MILLISECONDS); 
} 

private List<Data> generateData(){ 
    List<Data> dataList = new ArrayList<Data>(); 
    for (int i = 0; i < 20; i++) { 
     Data data = new Data("key"+i, "value"+i); 
     dataList.add(data); 
    } 
    return dataList; 
} 

Tôi bắt đầu thảo luận ở RxJava Google group nơi họ đã giúp tôi giải quyết.