聊聊reactor extra的retry

序

本文主要研究一下reactor extra的retry

maven

<dependency>
			<groupId>io.projectreactor.addons</groupId>
			<artifactId>reactor-extra</artifactId>
			<version>3.1.4.RELEASE</version>
		</dependency>

實例

TcpClient client = TcpClient.create("localhost", 8888);
        client.newHandler((inbound,outbound) -> {
                return outbound.sendString(Mono.just("Hello World!")).then(inbound.receive()
                        .asString().next().log().then());
        }).doOnError(e -> e.printStackTrace())
                .subscribe();

上面這個TcpClient，在server沒有啓動的狀況下鏈接會直接報錯

io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: localhost/127.0.0.1:8888
	at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
	at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
	at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:325)
	at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:340)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:633)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886)
	at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.ConnectException: Connection refused
	... 10 more

鏈接失敗重連

簡單重試

client.newHandler((inbound,outbound) -> {
                return outbound.sendString(Mono.just("Hello World!")).then(inbound.receive()
                        .asString().next().log().then());
        }).doOnError(e -> e.printStackTrace())
                .retry(3)
                .subscribe();

retry能夠直接指定重試次數

高級重試

client.newHandler((inbound,outbound) -> {
                return outbound.sendString(Mono.just("Hello World!")).then(inbound.receive()
                        .asString().next().log().then());
        }).doOnError(e -> e.printStackTrace())
                .retryWhen(Retry.allBut(RuntimeException.class)
                .retryMax(5000)
                .fixedBackoff(Duration.ofSeconds(5))
                .doOnRetry(e -> {
                    e.exception().printStackTrace();
                })
        )
                .subscribe();

利用reactor extra項目中的Retry幫助類，能夠輕鬆指定高級重試策略，好比fixedBackoff，亦或是exponentialBackoff等

client.newHandler((inbound,outbound) -> {
                return outbound.sendString(Mono.just("Hello World!")).then(inbound.receive()
                        .asString().next().log().then());
        }).doOnError(e -> e.printStackTrace())
                .retryWhen(Retry.allBut(RuntimeException.class)
                .retryMax(5000)           .exponentialBackoffWithJitter(Duration.ofMillis(100),Duration.ofMillis(500))
                .doOnRetry(e -> {
                    e.exception().printStackTrace();
                })
        )
                .subscribe();

這裏使用了exponentialBackoffWithJitter，第一個參數是firstBackoff時間，第二個參數是maxBackoff，也就是maxBackoffInterval，若是爲null則至關於Duration.ofSeconds(Long.MAX_VALUE)

Retry

reactor-extra-3.1.4.RELEASE-sources.jar!/reactor/retry/Retry.java

/**
	 * Returns a retry function that retries errors resulting from all exceptions except
	 * the specified non-retriable exceptions. More constraints may be added using
	 * {@link #retryMax(int)} or {@link #timeout(Duration)}.
	 *
	 * @param nonRetriableExceptions exceptions that may not be retried
	 * @return retry function that retries all exceptions except the specified non-retriable exceptions.
	 */
	@SafeVarargs
	static <T> Retry<T> allBut(final Class<? extends Throwable>... nonRetriableExceptions) {
		Predicate<? super RetryContext<T>> predicate = context -> {
			Throwable exception = context.exception();
			if (exception == null)
				return true;
			for (Class<? extends Throwable> clazz : nonRetriableExceptions) {
				if (clazz.isInstance(exception))
					return false;
			}
			return true;
		};
		return DefaultRetry.<T>create(predicate);
	}

能夠看到使用DefaultRetry來建立

reactor-extra-3.1.4.RELEASE-sources.jar!/reactor/retry/DefaultRetry.java

public static <T> DefaultRetry<T> create(Predicate<? super RetryContext<T>> retryPredicate) {
		return new DefaultRetry<T>(retryPredicate,
				1,
				null,
				Backoff.zero(),
				Jitter.noJitter(),
				null,
				NOOP_ON_RETRY,
				(T) null);
	}

注意這裏的maxIterations默認爲1，也就是若是不指定retryMax，至關於高級重試策略就白費了，這個要額外注意一下。

小結

Reactor Extra提供的Retry工具類很是好用，值得實驗一下。

doc

• Reactor Extra