1. 引言

最近遇到一个生活场景，需要把消息队列里的故障消息，推送给PC客户端，并在客户端主动语音播报。
这个功能涉及语音合成和通知推送，对于通知推送使用了WebSocket，下面来记录下。

2. WebSocket使用步骤

SpringBoot集成WebSocket

2.1 引入依赖

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-websocketartifactId>
dependency>
1
2
3
4

2.2 创建WebSocket配置类

/**
 * WebSocket配置
 */
@Configuration
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

}
1
2
3
4
5
6
7
8
9
10
11
12

WebSocket 推送实例

zip 3星 超过45%的资源 28KB

下载

2.3 WebSocket服务类

@Slf4j
@ServerEndpoint("/websocket/{userId}")
@Component
public class WebSocketServer {


    // 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;

    // concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = 
    										new CopyOnWriteArraySet<WebSocketServer>();

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    //接收userId
    private String userId="";

    public static String baseUrl = "http://118.89.68.236:9000";

    /**
     * 连接建立成功调用的方法*/
    @OnOpen
    public void onOpen(Session session,@PathParam("userId") String userId) {
        this.session = session;
        webSocketSet.add(this);     //加入set中
        addOnlineCount();           //在线数加1
        log.info("有新窗口开始监听:"+userId+",当前在线人数为" + getOnlineCount());
        this.userId = userId;
        try {
            sendMessage("连接成功");
        } catch (IOException e) {
            log.error("websocket IO异常");
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);  //从set中删除
        subOnlineCount();           //在线数减1
        log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     * @param message 客户端发送过来的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("收到来自窗口"+userId+"的信息:"+message);
        //群发消息
        for (WebSocketServer item : webSocketSet) {
            try {
                item.sendMessage(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误");
        error.printStackTrace();
    }
    /**
     * 实现服务器主动推送-字符串
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 实现服务器主动推送- byte数组
     */
    public void sendAudio(byte[] data) throws IOException {
        ByteBuffer buffer = ByteBuffer.wrap(data);
        this.session.getBasicRemote().sendBinary(buffer);
    }


    /**
     * 群发自定义消息-发送音频文件
     * */
    public static void sendToAudio(String message){
        InputStream inputStream = null;
        for (WebSocketServer item : webSocketSet) {
            try {
                inputStream = sendGet(message);
                byte[] bytes = IOUtils.toByteArray(inputStream);
                item.sendAudio(bytes);
            } catch (Exception e) {
            }
        }
    }

   
    /**
     * 群发自定义消息-发送文本文件
     * */
    public static void sendInfo(String message,@PathParam("userId") String userId){
        log.info("推送消息到窗口"+userId+"，推送内容:"+message);
        for (WebSocketServer item : webSocketSet) {
            try {
                //这里可以设定只推送给这个userId的，为null则全部推送
                if(userId==null) {
                    item.sendMessage(message);
                }else if(item.userId.equals(userId)){
                    item.sendMessage(message);
                }
            } catch (IOException e) {
                continue;
            }
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
    
    public static InputStream sendGet(String message) {
        InputStream inputStream = null;
        try {
            String url = encode(baseUrl,message);
            URL serverUrl = new URL(url);
            URLConnection connection = serverUrl.openConnection();
            //设置超时时间
            connection.setConnectTimeout(5000);
            connection.setReadTimeout(15000);
            connection.setRequestProperty("accept", "*/*");
            connection.setRequestProperty("connection", "Keep-Alive");
            connection.connect();
            inputStream = connection.getInputStream();
        } catch (Exception e){
            e.printStackTrace();
        } finally {

        }
        return inputStream;
    }

    public static String encode(String baseUrl, String message) throws UnsupportedEncodingException {
        String title = "format: yaml\nmode: mspk\naudio: 14\nspeaker: Aida\nvocoder: melgan";
        String encodeContent = URLEncoder.encode(message, "UTF-8");
        String encodeTitle = URLEncoder.encode(title, "UTF-8");
        String result = baseUrl + "/synthesize?text=" + encodeContent + "&kwargs=" + encodeTitle;
        log.info("request path : {}",result);
        return result;
    }

}
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

对于上面根据文字产生音频文件的两个方法：sendGet()、encode()。返回的是文件流，对于的服务使用的是ttskit，可以去GitHub搜一下。

通过使用IOUtils.toByteArray(inputStream)方法将InputStream转为byte[]数组，使用的commons-io包中的工具类。

<dependency>
    <groupId>commons-iogroupId>
    <artifactId>commons-ioartifactId>
    <version>2.6version>
dependency>
1
2
3
4
5

最后就可以在业务代码中，调用WebSocket服务方法了。此处使用定时任务模拟消息推送。

@Slf4j
@Configuration
@EnableScheduling
public class TTSTask {

    public static int i = 1;
    @Scheduled(cron = "0/30 * * * * ?")
    private void configureTasks() {
        log.info("执行静态定时任务时间: " + LocalDateTime.now());
        String message = "收到一条故障信息，南京市江北新区高新路"+ i + "号杆塔故障";
        log.info("播报信息“{}",message);
        WebSocketServer.sendToAudio(message);
        i++;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

2.4 前端页面

DOCTYPE html>
<html lang="en" xmlns:th="http://www.thymeleaf.org">
<head>
    <meta charset="UTF-8">
    <title>语音自动播报title>
    <script type="text/javascript" src="/js/jquery.js" th:src="@{/js/jquery.js}">script>
head>
<body>
<header>
    <h1 align="center">语音自动播报h1>
header>
<div>
    <input type="button" id="btnConnection" value="连接" />
    <input type="button" id="btnClose" value="关闭" />
div>
<div align="center" style="margin-top:10px">
    <audio id="audio" controls autoplay hidden>audio>
div>

<script type="text/javascript">
    $(function() {
        var socket;
        if(typeof(WebSocket) == "undefined") {
            alert("您的浏览器不支持WebSocket");
            return;
        }

        $("#btnConnection").click(function() {
            //实现化WebSocket对象，指定要连接的服务器地址与端口
            socket = new WebSocket("ws://127.00.1:8890/websocket/88888");
            //打开事件
            socket.onopen = function() {
                console.log("Socket 已打开");
            };
            //获得消息事件
            socket.onmessage = function(msg) {
                console.log("获得消息：",msg);
                console.log("获得消息：",msg.data);
                q('#audio').src = URL.createObjectURL(msg.data)
                q('#audio').hidden = false
            };
            //关闭事件
            socket.onclose = function() {
                console.log("Socket已关闭");
            };
            //发生了错误事件
            socket.onerror = function() {
                console.log("发生了错误");
            }
        });

        $("#btnClose").click(function() {
            socket.close();
        });
    });

	function q(selector) {
	    return document.querySelector(selector)
	}
script>
body>
html>
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

这里只说一点，对于 socket.onmessage方法的回调结果，对于byte[]数组，使用blob接收的，对于前端audio标签可以直接使用