MessageWebsocket.java
5.89 KB
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
package com.java110.api.websocket;
import com.alibaba.fastjson.JSONObject;
import com.java110.dto.WsDataDto;
import com.java110.utils.util.StringUtil;
import com.java110.vo.ResultVo;
import org.slf4j.Logger;
import com.java110.core.log.LoggerFactory;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
/**
* @ClassName MessageWebsocket
* @Description TODO
* @Author wuxw
* @Date 2020/6/5 12:16
* @Version 1.0
* add by wuxw 2020/6/5
**/
@ServerEndpoint("/ws/message/{userId}/{clientId}")
@Component
public class MessageWebsocket {
private final static Logger logger = LoggerFactory.getLogger(MessageWebsocket.class);
/**
* 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
*/
private static int onlineCount = 0;
/**
* concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
*/
private static ConcurrentHashMap<String, MessageWebsocket> webSocketMap = new ConcurrentHashMap<>();
/**
* concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
*/
private static ConcurrentHashMap<String, String> clientMachineMap = new ConcurrentHashMap<>();
/**
* 与某个客户端的连接会话，需要通过它来给客户端发送数据
*/
private Session session;
/**
* 接收clientId
*/
private String userId = "";
private String clientId = "";
/**
* 连接建立成功调用的方法
*/
@OnOpen
public void onOpen(Session session, @PathParam("userId") String userId, @PathParam("clientId") String clientId) {
this.session = session;
this.userId = userId;
this.clientId = clientId;
if (webSocketMap.containsKey(userId)) {
webSocketMap.remove(userId);
webSocketMap.put(userId, this);
//加入set中
} else {
webSocketMap.put(userId, this);
//加入set中
addOnlineCount();
//在线数加1
}
logger.debug("用户连接:" + userId + ",客户端：" + clientId + ",当前在线人数为:" + getOnlineCount());
try {
sendMessage("连接成功");
} catch (IOException e) {
logger.error("用户:" + userId + ",客户端：" + clientId + ",网络异常!!!!!!");
}
}
/**
* 连接关闭调用的方法
*/
@OnClose
public void onClose() {
if (webSocketMap.containsKey(userId)) {
webSocketMap.remove(userId);
//从set中删除
subOnlineCount();
}
logger.info("用户退出:" + userId + ",客户端：" + clientId + ",当前在线人数为:" + getOnlineCount());
}
/**
* 收到客户端消息后调用的方法
*
* @param message 客户端发送过来的消息
*/
@OnMessage
public void onMessage(String message, Session session) throws IOException {
logger.info("用户消息:" + userId + ",客户端：" + clientId + ",报文:" + message);
//可以群发消息
//消息保存到数据库、redis
if (StringUtil.isEmpty(message)) {
ResultVo resultVo = new ResultVo(ResultVo.CODE_ERROR, "未包含内容");
webSocketMap.get(userId).sendMessage(resultVo.toString());
return;
}
if (!StringUtil.isJsonObject(message)) {
ResultVo resultVo = new ResultVo(ResultVo.CODE_ERROR, "不是有效数据格式");
webSocketMap.get(userId).sendMessage(resultVo.toString());
return;
}
WsDataDto wsDataDto = JSONObject.parseObject(message, WsDataDto.class);
switch (wsDataDto.getCmd()) {
case WsDataDto.CMD_PING:
//webSocketMap.get(userId).sendMessage(wsDataDto.toString());
break;
}
// //解析发送的报文
// JSONObject jsonObject = JSONObject.parseObject(message);
// //追加发送人(防止串改)
// jsonObject.put("fromUserId", this.userId);
// String toUserId = jsonObject.getString("toUserId");
// //传送给对应toUserId用户的websocket
// if (!StringUtil.isEmpty(toUserId) && webSocketMap.containsKey(toUserId)) {
// webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());
// } else {
// logger.error("请求的clientId:" + toUserId + "不在该服务器上");
// //否则不在这个服务器上，发送到mysql或者redis
// }
}
/**
* @param session
* @param error
*/
@OnError
public void onError(Session session, Throwable error) {
logger.error("用户错误:" + this.userId + ",客户端：" + clientId + ",原因:" + error.getMessage());
// error.printStackTrace();
}
/**
* 实现服务器主动推送
*/
public void sendMessage(String message) throws IOException {
//this.session.getBasicRemote().sendText(message);
synchronized (this.session) {
this.session.getBasicRemote().sendText(message);
}
}
/**
* 发送设备监控信息
*/
public static void sendInfo(String message, String userId) throws IOException {
logger.info("发送消息到:" + userId + "，报文:" + message);
if (!webSocketMap.containsKey(userId)) {
//客户端未连接
return;
}
webSocketMap.get(userId).sendMessage(message);
}
public static synchronized int getOnlineCount() {
return onlineCount;
}
public static synchronized void addOnlineCount() {
MessageWebsocket.onlineCount++;
}
public static synchronized void subOnlineCount() {
MessageWebsocket.onlineCount--;
}
}