溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章給大家分享的是有關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進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
