spring-boot - 如何重新连接 ReactorNettyWebSocketClient 连接？

Question

我需要访问一个 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。但是如何重新连接套接字？

Answer 1

没有开箱即用的解决方案，重新连接机制不是 JSR 356 - WebSocket 的 Java API 的一部分。但是您可以自己实现它 - 一个带有 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());
}