java WebSocket客户端断线重连 | 实用代码框架「建议收藏」

java WebSocket客户端断线重连 | 实用代码框架「建议收藏」在工作中是否会遇到实用websocket客户端连接服务端的时候,网络波动,服务端断连的情况。会导致客户端被动断开连接。为了解决这个问题,需要对被动断开连接的情况进行捕获,并重新创建连接。这篇文章主要是提供可以直接使用的断线重连websocket客户端代码。

大家好,又见面了,我是你们的朋友全栈君。

目录

前言

Maven依赖

代码

总结


前言

在工作中是否会遇到实用websocket客户端连接服务端的时候,网络波动服务端断连的情况。会导致客户端被动断开连接。为了解决这个问题,需要对被动断开连接的情况进行捕获,并重新创建连接。这篇文章主要是提供可以直接使用的断线重连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;
/** @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客户端断线重连 | 实用代码框架「建议收藏」

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/157472.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(1)
blank

相关推荐

  • Alex 的 Hadoop 菜鸟教程: 第15课 Impala 安装使用教程

    Alex 的 Hadoop 菜鸟教程: 第15课 Impala 安装使用教程本教程介绍Impala的安装,使用和JDBC调用。为什么用Impala?因为Hive太慢了!Impala也可以执行SQL,但是比Hive的速度快很多。为什么Impala可以比Hive快呢?因为Hive采用的是把你的sql转化成hadoop的MapReduce任务的代码,然后编译,打包成jar包,并分发到各个server上执行,这是一个相当慢的过程。而Impala根本就不用Hadoop的MapReduce机制,直接调用HDFS的API获取文件,在自己的内存中进行计算。

  • Corners检测「建议收藏」

    Corners检测「建议收藏」HarrisCorners角点检测使用harriscorners检测器:使用OpenCV函数voidcornerHarris(InputArraysrc,OutputArraydst,intblockSize,                intksize,doublek,

    2022年10月22日
  • PAT乙级-坑点集合

    PAT乙级-坑点集合PAT乙级虽然挺简单但是坑不少。因为不给出测试数据,所以不知道怎么才能AC。以下几个题都是只做对了一部分,剩下的错误的点怎么想都不知道怎么才能过,所以整理一下,等回头再慢慢琢磨。1.1008数组元素循环右移问题1008数组元素循环左移2.1012数字分类1012数字分类3.1014福尔摩斯的约会1014福尔摩斯的约会4.1017A/B…

  • ssm整合思路与配置详解_接口整合配置

    ssm整合思路与配置详解_接口整合配置swagger2于17年停止维护,现在最新的版本为17年发布的Swagger3(OpenApi3)

    2022年10月30日
  • NSGA2算法及其代码

    NSGA2算法及其代码本人最近研究NSGA2算法,网上有很多示例代码,但是基本没有注释,代码看起来很头疼,因此我最近把整个代码研读了一遍,并做上中文注释,希望可以帮助到一些和我一样的初学者们。贴出代码之前,首先介绍一下NSGA2遗传算法的流程图:流程图中我把每个详细的步骤用号码标出来,对应下文的代码部分。首先贴出主函数代码,对应整个流程图:functionnsga_2_optimization%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%…

  • FPGA实现的SPI协议(二)—-基于SPI接口的FLASH芯片M25P16的使用「建议收藏」

    FPGA实现的SPI协议(二)—-基于SPI接口的FLASH芯片M25P16的使用「建议收藏」用flash芯片M25P16来验证下SPI的具体实现吧。

    2022年10月15日

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号