这篇文章给大家分享的是有关java中如何实现WebSocket客户端断线重连的内容。小编觉得挺实用的,因此分享给大家做个参考,一起跟随小编过来看看吧。
<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; /** @Author huyi @Date 2021/10/15 20:03 @Description: 重连websocket客户端 */ @Slf4j public class ReConnectWebSocketClient { /** 字符串消息回调 */ private Consumer<String> msgStr; /** 字节流消息回调 */ private Consumer<ByteBuffer> msgByte; /** 异常回调 */ private Consumer<Exception> error; /** 连接标识 */ private String key; /** ws服务端连接 */ private URI serverUri; /** 尝试重连标识 */ private AtomicBoolean tryReconnect; /** 需要ping标识 */ private AtomicBoolean needPing; /** websocket连接实体 */ 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(); } /** * 连接 * * @throws Exception 异常 */ public void connect() throws Exception { webSocketClient.connectBlocking(10, TimeUnit.SECONDS); } /** * 发送 * * @param msg 消息 * @throws Exception 异常 */ public void send(String msg) throws Exception { this.initReConnectReq = msg; if (webSocketClient.isOpen()) { webSocketClient.send(msg); } } /** * 关闭 * * @param msg 关闭消息 */ public void close(String msg) { needPing.set(false); end.set(true); if (webSocketClient != null) { webSocketClient.closeConnection(3, msg); } } /** * 忽略证书 * * @param client */ 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); } } }
代码说明:
1、参数的重连次数可以配置。
2、增加异步pingpong线程,一旦结束连接会自动关闭。
3、对字符串、字节流、异常都有回调措施。
测试代码方法
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"); }
验证结果
16: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客户端断线重连”这篇文章就分享到这里了,希望以上内容可以对大家有一定的帮助,让大家可以学到更多知识,如果觉得文章不错,可以把它分享出去让更多的人看到吧!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。