RxJava中的指數退避

Question

我有一個需要觸發事件的Observable的API。

我想返回Observable，如果檢測到Internet連接，則每defaultDelay秒發出一個值，如果沒有連接，則延遲numberOfFailedAttempts^2次。

我已經嘗試了一堆各種風格，我在被觀察到的retryWhen's只計算一次最大的問題：

Observable 
    .interval(defaultDelay,TimeUnit.MILLISECONDS) 
    .observeOn(Schedulers.io()) 
    .repeatWhen((observable) -> 
     observable.concatMap(repeatObservable -> { 
      if(internetConnectionDetector.isInternetConnected()){ 
       consecutiveRetries = 0; 
       return observable; 
      } else { 
       consecutiveRetries++; 
       int backoffDelay = (int)Math.pow(consecutiveRetries,2); 
       return observable.delay(backoffDelay, TimeUnit.SECONDS); 
       } 
     }).onBackpressureDrop()) 
    .onBackpressureDrop(); 

有沒有辦法做什麼，我試圖做的？我發現了一個相關的問題（目前找不到它），但採取的方法似乎不適用於動態值。

Answer 1

在代碼中有兩處錯誤：

1. 爲了重複一些可觀察序列，該序列必須是有限的。即而不是像interval那樣，您最好使用像just或fromCallable之類的東西，就像我在下面的示例中所做的那樣。
2. 從repeatWhen的內部函數您需要返回新的延遲觀察源，所以而不是observable.delay()你必須返回Observable.timer()。

工作代碼：

public void testRepeat() throws InterruptedException { 
    logger.info("test start"); 

    int DEFAULT_DELAY = 100; // ms 
    int ADDITIONAL_DELAY = 100; // ms 
    AtomicInteger generator = new AtomicInteger(0); 
    AtomicBoolean connectionAlive = new AtomicBoolean(true); // initially alive 

    Disposable subscription = Observable.fromCallable(generator::incrementAndGet) 
      .repeatWhen(counts -> { 
       AtomicInteger retryCounter = new AtomicInteger(0); 
       return counts.flatMap(c -> { 
        int retry = 0; 
        if (connectionAlive.get()) { 
         retryCounter.set(0); // reset counter 
        } else { 
         retry = retryCounter.incrementAndGet(); 
        } 
        int additionalDelay = ADDITIONAL_DELAY * (int) Math.pow(retry, 2); 
        logger.info("retry={}, additionalDelay={}ms", retry, additionalDelay); 
        return Observable.timer(DEFAULT_DELAY + additionalDelay, TimeUnit.MILLISECONDS); 
       }); 
      }) 
      .subscribe(v -> logger.info("got {}", v)); 

    Thread.sleep(220); 
    logger.info("connection dropped"); 
    connectionAlive.set(false); 
    Thread.sleep(2000); 
    logger.info("connection is back alive"); 
    connectionAlive.set(true); 
    Thread.sleep(2000); 
    subscription.dispose(); 
    logger.info("test complete"); 
} 

查看關於repeatWhenhere詳細的文章。

Answer 2

您可以使用retryWhen運算符來配置沒有連接時的延遲。如何定期發出物品是一個單獨的主題（查找interval或timer運營商）。如果你無法弄清楚，請另外打個問題。

我在我的Github上有一個廣泛的例子，但我會給你這裏的要點。

RetryWithDelay retryWithDelay = RetryWithDelay.builder() 
    .retryDelayStrategy(RetryDelayStrategy.RETRY_COUNT) 
    .build() 

Single.fromCallable(() -> { 
    ... 
}).retryWhen(retryWithDelay) 
.subscribe(j -> { 
    ... 
}) 

RetryWithDelay定義如下。我使用了RxJava 2.x，所以如果你使用1.x，簽名應該是Func1<Observable<? extends Throwable>, Observable<Object>>。

public class RetryWithDelay implements 
     Function<Flowable<? extends Throwable>, Publisher<Object>> { 
    ... 
} 

RetryWithDelay class。

RetryStrategy enum。

這使我可以根據RetryDelayStrategy配置各種類型的超時，常量，線性，指數。對於您的使用案例，您應選擇CONSTANT_DELAY_TIMES_RETRY_COUNT延遲策略，並在編制RetryWithDelay時致電retryDelaySeconds(2)。

retryWhen是一個複雜的，甚至可能是越野車的操作員。你可以在網上找到的大多數例子都使用range操作符，如果沒有重試就會失敗。有關詳細信息，請參閱我的回答here。

Answer 3

我一直髮現retryWhen有些低級別，所以對於指數回退，我使用一個構建器（如Abhijit），它是單元測試的，可用於RxJava 1.x的rxjava-extras。我建議使用加蓋版本，以便延遲的指數增加不會超出您定義的最大值。

這是你如何使用它：

observable.retryWhen(
    RetryWhen.exponentialBackoff(
     delay, maxDelay, TimeUNIT.SECONDS) 
    .build()); 

我不同意retryWhen是馬車，但如果你發現它RxJava的bug報告。 Bug快速修復！

你需要 rxjava-額外 0.8.0.6或更高版本是Maven的中央：

<dependency> 
    <groupId>com.github.davidmoten</groupId> 
    <artifactId>rxjava-extras</artifactId> 
    <version>0.8.0.6</version> 
</dependency> 

讓我知道如果你需要的RxJava 2.x版。從0.1.4開始，rxjava2-extras中提供了相同的功能。