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

• 使用演示

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

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

基于Springboot工程

• 引入websocket依赖

	<!--websocket-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>
1
2
3
4
5

• 开箱即用的工具类

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() {
                    @Override
                    public 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, "正常关闭"));
    }

    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
    }

    /**
     * 接收消息,接收消息的响应动作由使用者在send的时候自行传入
     *
     * @param message 消息内容
     */
    @OnMessage
    public void onMessage(String message) {
        receiveConsumer.accept(unicodeDecode(message));
    }

    /**
     * 关闭时的操作，分为正常关闭和异常关闭，这里异常关闭的做重连操作
     */
    @OnClose
    public 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);
        }
    }

    @OnError
    public 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;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212