32

Flux 内の要素をマッピングするときにエラーを処理する方法を理解しようとしています。

たとえば、CSV 文字列をビジネス POJO の 1 つに解析しています。

myflux.map(stock -> converter.convertHistoricalCSVToStockQuotation(stock));

この行の一部にエラーが含まれている可能性があるため、ログには次のように表示されます。

 reactor.core.publisher.FluxLog:  onNext([SOME_BOGUS_QUOTE]@38.09 (Fri Apr 08 00:00:00 CEST 2016) H(38.419998)/L(37.849998)/O(37.970001))
 reactor.core.publisher.FluxLog:  onNext([SOME_BOGUS_QUOTE]@38.130001 (Thu Apr 07 00:00:00 CEST 2016) H(38.189999)/L(37.610001)/O(37.799999))
 reactor.core.publisher.FluxLog:  onError(java.lang.IllegalArgumentException: Invalid CSV stock quotation: SOME_BOGUS_QUOTE,trololo)
 reactor.core.publisher.FluxLog:  java.lang.IllegalArgumentException: Invalid CSV stock quotation: SOME_BOGUS_QUOTE,trololo

API でいくつかのエラー処理メソッドを読みましたが、ほとんどは「エラー値」を返すか、次のようなフォールバック Flux を使用することを参照していました。

Flux.onErrorResumeWith(myflux, x -> Mono.fromCallable(() -> ... do stuff);

ただし、これを my で使用myfluxすると、フラックス全体が再度処理されます。

では、特定の要素の処理中にエラーを処理し (つまり、それらを無視する/ログに記録する)、残りのフラックスの処理を続ける方法はありますか?

@akarnokd 回避策で更新

public Flux<StockQuotation> getQuotes(List<String> tickers)
{
    Flux<StockQuotation> processingFlux = Flux.fromIterable(tickers)
    // Get each set of quotes in a separate thread
    .flatMap(s -> Mono.fromCallable(() -> feeder.getCSVQuotes(s)))
    // Convert each list of raw quotes string in a new Flux<String>
    .flatMap(list -> Flux.fromIterable(list))
    // Convert the string to POJOs
    .flatMap(x -> {
            try {
                return Flux.just(converter.convertHistoricalCSVToStockQuotation(x));    
            }
            catch (IllegalArgumentException ex){
                System.out.println("Error decoding stock quotation: " + x);
                return Flux.empty();
            }
    });

    return processingFlux;
}

ただし、コードが以前よりも洗練されていないことがわかります。Flux API には、このコードが行うことを行うメソッドがありませんか?

retry(...)
retryWhen(...)
onErrorResumeWith(...)
onErrorReturn(...)
4

5 に答える 5

31

flatMap代わりに、処理が失敗した場合に空のシーケンスを返す必要があります。

myflux.flatMap(v -> {
    try {
        return Flux.just(converter.convertHistoricalCSVToStockQuotation(stock));
    } catch (IllegalArgumentException ex) {
        return Flux.empty();
    }
});
于 2016-03-30T10:23:27.970 に答える