如何重新连接 ReactorNettyWebSocketClient 连接？答案

【问题标题】：How to reconnect ReactorNettyWebSocketClient connection?如何重新连接 ReactorNettyWebSocketClient 连接？
【发布时间】：2018-07-13 21:01:52
【问题描述】：

我需要访问一个 websocket 服务，它会在 24 小时后关闭一个打开的 websocket 连接。如何使用 Spring-Boot 2 和 Webflux 实现重新连接？

这是我目前所拥有的（取自https://github.com/artembilan/webflux-websocket-demo）：

@GetMapping(path = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> getStreaming() throws URISyntaxException {
    ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient();

    EmitterProcessor<String> output = EmitterProcessor.create();

    Mono<Void> sessionMono = client.execute(new URI("ws://localhost:8080/echo"),

    session -> session.receive()
    .timeout(Duration.ofSeconds(3))
    .map(WebSocketMessage::getPayloadAsText)
    .subscribeWith(output)
    .then());

    return output.doOnSubscribe(s -> sessionMono.subscribe());
}

一旦连接丢失（3 秒内没有输入），就会抛出 TimeoutException。但是如何重新连接套接字？

【问题讨论】：

标签： spring-boot spring-webflux reactor-netty

【解决方案1】：

没有开箱即用的解决方案，重新连接机制不是 JSR 356 - Java API for WebSocket 的一部分。但是您可以自己实现它 - 一个简单的 Spring 事件示例：

第 1 步 - 创建事件类。

public class ReconnectionEvent extends ApplicationEvent {

    private String url;

    public ReconnectionEvent(String url) {
        super(url);
        this.url = url;
    }

    public String getUrl() {
        return url;
    }

}

第 2 步 - 提供 websocket 连接方法。 示例：

...
@Autowired
private ApplicationEventPublisher publisher;
...
public void connect(String url) {

    ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient();

    EmitterProcessor<String> output = EmitterProcessor.create();

    Mono<Void> sessionMono = client.execute(URI.create(url),

            session -> session.receive()
                    .map(WebSocketMessage::getPayloadAsText)
                    .log()
                    .subscribeWith(output)
                    .doOnTerminate(() -> publisher.publishEvent(new ReconnectEvent(url)))
                    .then());

    output
            .doOnSubscribe(s -> sessionMono.subscribe())
            .subscribe();
}

检查doOnTerminate() 方法 - 当 Flux 终止时，无论是成功完成还是出错，它都会发出一个 ReconnectEvent。如有必要，您可以在其他 Flux 的回调上发出重新连接事件（例如仅在 doOnError() 上）。

第 3 步 - 提供一个侦听器，当重新连接事件发生时，它会在给定的 url 上再次连接。

@EventListener(ReconnectEvent.class)
public void onApplicationEvent(ReconnectEvent event) {
    connect(event.getUrl());
}

【讨论】：

嗨@joanna，非常感谢 - 我会尽快查看！
不能将doOnTerminate 简化为() -> connect(url) 吗？需要发出应用程序事件的原因是什么？

【解决方案2】：

我用UnicastProcessor的reactor做了一些事情。

...
public abstract class AbstractWsReconnectClient {
    private Logger ...

    protected UnicastProcessor<AbstractWsReconnectClient> reconnectProcessor = UnicastProcessor.create();

    protected AbstractWsReconnectClient(Duration reconnectDuration) {
        reconnect(reconnectDuration);
    }

    public abstract Mono<Void> connect();

    private void reconnect(Duration duration) {
        reconnectProcessor.publish()
            .autoConnect()
            .delayElements(duration)
            .flatMap(AbstractWsReconnectClient::connect)
            .onErrorContinue(throwable -> true,
                (throwable, o) -> {
                    if (throwable instanceof ConnectException) {
                        logger.warn(throwable.getMessage());
                    } else {
                        logger.error("unexpected error occur during websocket reconnect");
                        logger.error(throwable.getMessage());
                    }
                })
            .doOnTerminate(() -> logger.error("websocket reconnect processor terminate "))
            .subscribe();
    }
}

当WebSocketClient终止时，调用UnicastProcessor.onNext

public Mono<Void> connect() {
    WebSocketClient client = new ReactorNettyWebSocketClient();
    logger.info("trying to connect to sso server {}", uri.toString());
    return client.execute(uri, headers, ssoClientHandler)
        .doOnTerminate(() -> {
            logger.warn("sso server {} disconnect", uri.toString());
            super.reconnectProcessor.onNext(this);
        });
}

【讨论】：