Springboot 集成WebSocket作为客户端，含重连接功能，开箱即用

• 使用演示

 public static void main(String[] args) throws Exception{//初始化socket客户端BaseWebSocketClient socketClient = BaseWebSocketClient.init("传入链接");//发送消息socketClient.sendMessage("填写需要发送的消息", (receive) -> {//这里编写接收消息的代码});}

只需要init后调用sendMessage方法即可，做到开箱即用。内部封装了失败重连接、断线重连接等功能。

基于Springboot工程

• 引入websocket依赖

	<!--websocket--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId></dependency>

• 开箱即用的工具类

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;import javax.websocket.ClientEndpoint;
import javax.websocket.CloseReason;
import javax.websocket.ContainerProvider;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.WebSocketContainer;
import java.io.IOException;
import java.net.URI;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;/*** 封装简易websocket客户端的类，带重连机制*/
@ClientEndpoint
@Slf4j
public class BaseWebSocketClient {/*** 接收消息的函数,发送消息的时候传入*/private Consumer<String> receiveConsumer;/*** 连接socket的url,init的时候传入*/private String url;/*** 当前socket会话对象，init执行后生成*/private Session session;/*** 重连延迟2.5秒执行（连接需要时间，重连的时候延迟执行）*/private final Long reconnectTime = 2500L;/*** 重连次数*/private AtomicInteger retryReconnectCount = new AtomicInteger(0);/*** 发送消息重试次数*/private AtomicInteger reconnectSendCount = new AtomicInteger(0);/*** 发送消息最大重试次数*/private final int maxReconnectSendCount = 10;/*** 初始化，初始化完才能正常使用** @param url websocket连接的地址*/public static BaseWebSocketClient init(String url) throws Exception {BaseWebSocketClient client = new BaseWebSocketClient();URI uri = new URI(url);WebSocketContainer container = ContainerProvider.getWebSocketContainer();container.connectToServer(client, uri);client.setUrl(url);return client;}/*** 发送消息** @param message         消息* @param receiveConsumer 接收消息的函数*/public void sendMessage(String message, Consumer<String> receiveConsumer) {if (session == null) {throw new RuntimeException("socket还未初始化");}this.setReceiveConsumer(receiveConsumer);try {if (session.isOpen()) {//如果是open状态就能够发送消息session.getBasicRemote().sendText(message);reconnectSendCount = new AtomicInteger(0);} else {//进行重连this.reconnect();//重连2s后重新发送消息new Timer().schedule((new TimerTask() {@Overridepublic void run() {//为了防止重试次数过多，这里做一下限制，一直连接不成功的就不发消息了if (reconnectSendCount.getAndIncrement() >= maxReconnectSendCount) {return;}//再次重试发送消息sendMessage(message, receiveConsumer);}}), reconnectTime + reconnectTime);}} catch (Exception e) {log.error("socket发送消息失败，url:{}", url, e);}}/*** 手动关闭连接，当不使用的时候手动关闭，减少连接资源的损耗*/public void close() throws IOException {session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "正常关闭"));}@OnOpenpublic void onOpen(Session session) {this.session = session;}/*** 接收消息,接收消息的响应动作由使用者在send的时候自行传入** @param message 消息内容*/@OnMessagepublic void onMessage(String message) {receiveConsumer.accept(unicodeDecode(message));}/*** 关闭时的操作，分为正常关闭和异常关闭，这里异常关闭的做重连操作*/@OnClosepublic void onClose(CloseReason closeStatus) throws Exception {if (closeStatus == null || closeStatus.getCloseCode() != CloseReason.CloseCodes.NORMAL_CLOSURE) {log.info("socket连接异常关闭url:{},closeStatus:{}", closeStatus, url);//重连reconnect();} else {log.info("socket连接关闭:{}", url);}}@OnErrorpublic void onError(Throwable throwable) throws Exception {log.error("socket连接异常,url:{}", url, throwable);//重连reconnect();}/*** 重连机制*/private void reconnect() throws Exception {if (session == null || session.isOpen()) {return;}//schedule里的this不是当前client对象Object that = this;new Timer().schedule(new TimerTask() {public void run() {//如果是打开的就不执行重连if (session.isOpen()) {return;}log.info("当前socket重连次数:{},url:{}", retryReconnectCount.getAndIncrement(), url);try {URI uri = new URI(url);WebSocketContainer container = ContainerProvider.getWebSocketContainer();container.connectToServer(that, uri);retryReconnectCount = new AtomicInteger(0);log.info("重连成功");} catch (Exception e) {log.error("socket重连失败,url:{}", url, e);}}}, reconnectTime);}private void setReceiveConsumer(Consumer<String> receiveConsumer) {this.receiveConsumer = receiveConsumer;}private void setUrl(String url) {this.url = url;}/*** unicode转中文*/public static String unicodeDecode(String string) {if (StringUtils.isBlank(string)) {return string;}Pattern pattern = Pattern.compile("(\\\\u(\\p{XDigit}{4}))");Matcher matcher = pattern.matcher(string);char ch;while (matcher.find()) {ch = (char) Integer.parseInt(matcher.group(2), 16);string = string.replace(matcher.group(1), String.valueOf(ch));}return string;}
}