Web Uygulamalarındaki RxJava Gözlemlerini Kullanarak Performans İyileştirme Yetersizliği

Question

Gözlemleyici API'leri Gözlemlenebilirleri temel alarak, gelenekselleri engellemek yerine, gerçek bir avantaj olup olmadığını değerlendirmek için bazı testler yapıyorum.

• iyi: engelleme işlemlerini sarar bir Callable/DeferredResult dönmek DİNLENME Hizmetleri

  bütün örnek available on Githug

  Şaşırtıcı sonuçlar thoughput sonuçları olduğunu göstermektedir olduğunu.

• Bu kadar kötü değil: REST Hizmetlerini Engelleme.

• kötü: Sonuç RxJava gözlemlenebilir tarafından ayarlanan bir DeferredResult dönmek DİNLENME Hizmetleri.

  Uygulama:

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

Bu

benim Bahar WebApp olduğunu

@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: Ham çağrılabilir ve Gözlenebilen uç noktaları

@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; 
    } 
} 

hem Tam ben servis yanıt süresini artırmak için Thread.sleep(500) gecikme belirledik.

Async çağrılabilir ile: 700 rps'de

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

Engelleme hiçbir hata: Yük testlerinden Orada sonuçları

vardır etrafında 404 rps ama hataları

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

üretir

Async with Gözlemlenebilir: fazla 20 rps ve alır hataları er

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

10 bir corePoolSize ile Gözlenebilir yürüttüğü, ancak 50 artırarak de bir şey düzelmedi.

Açıklama ne olabilir?

UPDATE: Akarnokd tarafından önerildiği gibi, aşağıdaki değişiklikleri yaptım. Object.create öğesinden Object.fromCallable öğesine taşındı ve Scheduler'ı denetleyicide yeniden kullandı, ancak yine de aynı sonuçları alıyorum.

Answer 1

Sorun bir noktada programlama hatası nedeniyle ortaya çıktı. Aslında sorudaki örnek mükemmel çalışıyor.

Başkalarının sorun yaşamasını önlemek için bir uyarı: Observable.just(func)'u kullanmamaya dikkat edin, func aslında Gözlemlenebilir oluşturmada çağrılır. Yani herhangi Thread.sleep orada çağıran iş parçacığı

@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; 
} 

blok onlar beni work it out yardımcı nerede RxJava Google group bir tartışma başladı olacaktır yerleştirilir.