java中如何实现WebSocket客户端断线重连

这篇文章给大家分享的是有关java中如何实现WebSocket客户端断线重连的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。

Maven依赖

<dependency>            <groupId>org.projectlombok</groupId>            <artifactId>lombok</artifactId>            <optional>true</optional>        </dependency>        <dependency>            <groupId>cn.hutool</groupId>            <artifactId>hutool-all</artifactId>            <version>5.5.2</version>        </dependency>        <dependency>            <groupId>org.java-websocket</groupId>            <artifactId>Java-WebSocket</artifactId>            <version>1.5.1</version>        </dependency>

代码

不废话，上代码。

package ai.guiji.csdn.ws.client; import cn.hutool.core.thread.ThreadUtil;import cn.hutool.core.util.StrUtil;import lombok.extern.slf4j.Slf4j;import org.java_websocket.WebSocket;import org.java_websocket.client.WebSocketClient;import org.java_websocket.framing.Framedata;import org.java_websocket.handshake.ServerHandshake; import javax.net.ssl.*;import java.net.Socket;import java.net.URI;import java.nio.ByteBuffer;import java.security.cert.CertificateException;import java.security.cert.X509Certificate;import java.util.concurrent.TimeUnit;import java.util.concurrent.atomic.AtomicBoolean;import java.util.concurrent.atomic.AtomicInteger;import java.util.function.Consumer; @Slf4jpublic class ReConnectWebSocketClient {    private Consumer<String> msgStr;    private Consumer<ByteBuffer> msgByte;    private Consumer<Exception> error;    private String key;    private URI serverUri;    private AtomicBoolean tryReconnect;    private AtomicBoolean needPing;    private WebSocketClient webSocketClient;    private AtomicInteger reConnectTimes;    private AtomicBoolean end;    private String initReConnectReq;    private Consumer<String> endConsumer;   public ReConnectWebSocketClient(      URI serverUri,      String key,      Consumer<String> msgStr,      Consumer<ByteBuffer> msgByte,      Consumer<Exception> error) {    this.msgStr = msgStr;    this.msgByte = msgByte;    this.error = error;    this.key = key;    this.serverUri = serverUri;    this.tryReconnect = new AtomicBoolean(false);    this.needPing = new AtomicBoolean(true);    this.reConnectTimes = new AtomicInteger(0);    this.end = new AtomicBoolean(false);    this.endConsumer = this::close;    init();  }     public void init() {    // 创建连接    createWebSocketClient();    // ping线程    circlePing();  }   private void needReconnect() throws Exception {    ThreadUtil.sleep(10, TimeUnit.SECONDS);    int cul = reConnectTimes.incrementAndGet();    if (cul > 3) {      close("real stop");      throw new Exception("服务端断连，3次重连均失败");    }    log.warn("[{}]第[{}]次断开重连", key, cul);    if (tryReconnect.get()) {      log.error("[{}]第[{}]次断开重连结果 -> 连接正在重连，本次重连请求放弃", key, cul);      needReconnect();      return;    }    try {      tryReconnect.set(true);       if (webSocketClient.isOpen()) {        log.warn("[{}]第[{}]次断开重连，关闭旧连接", key, cul);        webSocketClient.closeConnection(2, "reconnect stop");      }      webSocketClient = null;      createWebSocketClient();      connect();      if (StrUtil.hasBlank(initReConnectReq)) {        send(initReConnectReq);      }    } catch (Exception exception) {      log.error("[{}]第[{}]次断开重连结果 -> 连接正在重连，重连异常:[{}]", key, cul, exception.getMessage());      needReconnect();    } finally {      tryReconnect.set(false);    }  }   private void createWebSocketClient() {    webSocketClient =        new WebSocketClient(serverUri) {          @Override          public void onOpen(ServerHandshake serverHandshake) {            log.info("[{}]ReConnectWebSocketClient [onOpen]连接成功{}", key, getRemoteSocketAddress());            tryReconnect.set(false);          }           @Override          public void onMessage(String text) {            log.info("[{}]ReConnectWebSocketClient [onMessage]接收到服务端数据：text={}", key, text);            msgStr.accept(text);          }           @Override          public void onMessage(ByteBuffer bytes) {            log.info("[{}]ReConnectWebSocketClient [onMessage]接收到服务端数据：bytes={}", key, bytes);            msgByte.accept(bytes);          }           @Override          public void onWebsocketPong(WebSocket conn, Framedata f) {            log.info(                "[{}]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode={}",                key,                f.getOpcode());          }           @Override          public void onClose(int i, String s, boolean b) {            log.info("[{}]ReConnectWebSocketClient [onClose]关闭，s={}，b={}", key, s, b);            if (StrUtil.hasBlank(s) || s.contains("https")) {              if (end.get()) {                return;              }              try {                needReconnect();              } catch (Exception exception) {                endConsumer.accept("reconnect error");                error.accept(exception);              }            }          }           @Override          public void onError(Exception e) {            log.info("[{}]ReConnectWebSocketClient [onError]异常，e={}", key, e);            endConsumer.accept("error close");            error.accept(e);          }        };    if (serverUri.toString().contains("wss://")) {      trustAllHosts(webSocketClient);    }  }   public void circlePing() {    new Thread(            () -> {              while (needPing.get()) {                if (webSocketClient.isOpen()) {                  webSocketClient.sendPing();                }                ThreadUtil.sleep(5, TimeUnit.SECONDS);              }              log.warn("[{}]Ping循环关闭", key);            })        .start();  }     public void connect() throws Exception {    webSocketClient.connectBlocking(10, TimeUnit.SECONDS);  }     public void send(String msg) throws Exception {    this.initReConnectReq = msg;    if (webSocketClient.isOpen()) {      webSocketClient.send(msg);    }  }     public void close(String msg) {    needPing.set(false);    end.set(true);    if (webSocketClient != null) {      webSocketClient.closeConnection(3, msg);    }  }     public void trustAllHosts(WebSocketClient client) {    TrustManager[] trustAllCerts =        new TrustManager[] {          new X509ExtendedTrustManager() {             @Override            public void checkClientTrusted(                X509Certificate[] x509Certificates, String s, Socket socket)                throws CertificateException {}             @Override            public void checkServerTrusted(                X509Certificate[] x509Certificates, String s, Socket socket)                throws CertificateException {}             @Override            public void checkClientTrusted(                X509Certificate[] x509Certificates, String s, SSLEngine sslEngine)                throws CertificateException {}             @Override            public void checkServerTrusted(                X509Certificate[] x509Certificates, String s, SSLEngine sslEngine)                throws CertificateException {}             @Override            public void checkClientTrusted(X509Certificate[] x509Certificates, String s)                throws CertificateException {}             @Override            public void checkServerTrusted(X509Certificate[] x509Certificates, String s)                throws CertificateException {}             @Override            public X509Certificate[] getAcceptedIssuers() {              return null;            }          }        };     try {      SSLContext ssl = SSLContext.getInstance("SSL");      ssl.init(null, trustAllCerts, new java.security.SecureRandom());      SSLSocketFactory socketFactory = ssl.getSocketFactory();      client.setSocketFactory(socketFactory);    } catch (Exception e) {      log.error("ReConnectWebSocketClient trustAllHosts 异常，e={0}", e);    }  }}

代码说明：

参数的重连次数可以配置。

增加异步pingpong线程，一旦结束连接会自动关闭。

对字符串、字节流、异常都有回调措施。

测试代码方法

  public static void main(String[] args) throws Exception {    ReConnectWebSocketClient client =        new ReConnectWebSocketClient(            new URI(String.format("wss://192.168.1.77:24009")),            "test",            // 字符串消息处理            msg -> {              // todo 字符串消息处理              System.out.println("字符串消息:" + msg);            },            null,            // 异常回调            error -> {              // todo 字符串消息处理              System.out.println("异常:" + error.getMessage());            });    client.connect();    client.send("haha");  }

验证结果

08:54.468 [WebSocketConnectReadThread-12] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onOpen]连接成功/192.168.1.77:24009
16:08:54.475 [WebSocketConnectReadThread-12] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onMessage]接收到服务端数据：text=connect success from tcp4:192.168.6.63:11018!
字符串消息:connect success from tcp4:192.168.6.63:11018!
16:08:56.080 [WebSocketConnectReadThread-12] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onClose]关闭，s=，b=true
16:09:06.097 [WebSocketConnectReadThread-12] WARN ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]第[1]次断开重连
16:09:06.150 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onOpen]连接成功/192.168.1.77:24009
16:09:06.150 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onMessage]接收到服务端数据：text=connect success from tcp4:192.168.6.63:11038!
字符串消息:connect success from tcp4:192.168.6.63:11038!
16:09:09.369 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:14.370 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:19.371 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:24.379 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:29.382 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:34.398 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:39.402 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:44.404 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:49.415 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:54.429 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:09:59.437 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:10:04.449 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:10:06.154 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:10:09.455 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:10:14.462 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:10:19.468 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onWebsocketPong]接收到服务端数据：opcode=PONG
16:10:19.644 [WebSocketConnectReadThread-16] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onClose]关闭，s=，b=true
16:10:29.654 [WebSocketConnectReadThread-16] WARN ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]第[2]次断开重连
16:10:31.710 [WebSocketConnectReadThread-19] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onError]异常，e={}
java.net.ConnectException: Connection refused: connect
 at java.net.DualStackPlainSocketImpl.connect0(Native Method)
 at java.net.DualStackPlainSocketImpl.socketConnect(DualStackPlainSocketImpl.java:79)
 at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
 at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
 at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
 at java.net.PlainSocketImpl.connect(PlainSocketImpl.java:172)
 at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
 at java.net.Socket.connect(Socket.java:589)
 at sun.security.ssl.SSLSocketImpl.connect(SSLSocketImpl.java:673)
 at org.java_websocket.client.WebSocketClient.run(WebSocketClient.java:461)
 at java.lang.Thread.run(Thread.java:748)
16:10:31.710 [WebSocketConnectReadThread-19] INFO ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]ReConnectWebSocketClient [onClose]关闭，s=error close，b=false
异常:Connection refused: connect
16:10:34.473 [Thread-0] WARN ai.guiji.csdn.ws.client.ReConnectWebSocketClient - [test]Ping循环关闭

这里我才用的是手动关闭服务端方式触发，客户端被动断连情况。重连两次，第二次服务端还未启动导致异常触发。

感谢各位的阅读！关于“java中如何实现WebSocket客户端断线重连”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，让大家可以学到更多知识，如果觉得文章不错，可以把它分享出去让更多的人看到吧！