SpringBoot 整合 websocket （六）—— 后端补充

前面的几篇文章将原始socketweb做了前后端封装，在实际项目中使用时如果是单对单的通讯已经是可以使用了，如果是在非单对单使用时，还是有一定问题的，例如有这样一个考试场景，当所有考生打开问卷时，每个考生在页面上的每一次操作都会实时发送到管理端的监控页面上，如每个考生正在做的题目序号，考生离开考试页面，切屏操作等。这个时候如果管理端的监控页面只在一个浏览器上的一个标签页里打开，前面所写就可以实现，但是如果需要在多个地方同时监控，那就有问题了，操作记录只会在一个页面上显示。
下面还是以前面的对话聊天为例。

打开3个浏览器

打开google,firefox,edge 3个浏览器，访问socket.html，
google 浏览器中的fromId 填写 zhangsan ; toId 填写 lisi 。
firefox和edge浏览器中的fromId 填写 lisi; toId 填写 zhangsan 。
在google中的对话框中填写“李四，你好，我是张三”，然后点击发送按钮后，在firefox和edge浏览器中查看，只会在一个浏览器中会显示上面的消息。如下图
而因为有心跳监测和重连功能，每次发送消息后，消息会显示在最近一次握手成功的浏览器上，例如下图的信息显示在firefox中。

实现聊天室功能

聊天室功能就是大家的消息所有人都能看到，也即上面的张三发送给李四的消息需要在2个浏览器中同时显示。
先上效果图
1
2

从图中的时间可以看到，firefox,edge同时收到了消息。
实现上面的功能其实比较简单，wocket通讯的原理就是建立通信通道，而每个通道都是独立，而前面的写法是用各自的fromId做通道ID，而firefox,edge同时打开同一个连接时设置的fromId相同时，那么通道ID也是相同的，所以就会导致发送消息时只会有一个浏览器能收到。而为了解决这个问题其实也很简单，只要保证每个通道的ID唯一即可。我们只要在WebSocketServer里的ConcurrentHashMap sessionPools的key做唯一性处理即可。

@Component
@ServerEndpoint("/webSocket/{id}")
public class WebSocketServer {

	private static final long sessionTimeout = 1000 * 60 * 10;
	// 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
	private static AtomicInteger onlineNum = new AtomicInteger();

	// concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。
	private static ConcurrentHashMap<String, Session> sessionPools = new ConcurrentHashMap<>();
	private static String socketId;

	// 建立连接成功调用
	@OnOpen
	public void onOpen(final Session session, @PathParam(value = "id") final String id) {
		session.setMaxIdleTimeout(WebSocketServer.sessionTimeout);
		try {
			synchronized (WebSocketServer.sessionPools) {
				WebSocketServer.socketId = id + "_" + session.getId();
				if (WebSocketServer.sessionPools.containsKey(WebSocketServer.socketId)) {
					WebSocketServer.sessionPools.remove(WebSocketServer.socketId);
				} else {
					WebSocketServer.addOnlineCount();
				}
				WebSocketServer.sessionPools.put(WebSocketServer.socketId, session);
			}
			System.out.println(id + "加入webSocket！当前人数为" + WebSocketServer.onlineNum);
			final WebSocketMsgVO messageVO = new WebSocketMsgVO();
			messageVO.setToId(id);
			messageVO.setContent("欢迎" + id + "加入连接！");
			messageVO.setSendMsgType("open");
			this.sendMessage(session, HnJsonUtils.jsonToString(messageVO));
		} catch (final IOException e) {
			e.printStackTrace();
		}
	}

	// 关闭连接时调用
	@OnClose
	public void onClose(final Session session, @PathParam(value = "id") final String id) {
		final String sId = id + "_" + session.getId();
		synchronized (WebSocketServer.sessionPools) {
			if (WebSocketServer.sessionPools.containsKey(sId)) {
				WebSocketServer.sessionPools.remove(sId);
				WebSocketServer.subOnlineCount();
			}
		}
		System.out.println(sId + "断开webSocket连接！当前人数为" + WebSocketServer.onlineNum + " ** "
				+ HnDateUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
	}

	// 收到客户端信息
	@OnMessage
	public void onMessage(final String message, @PathParam(value = "id") final String id) throws IOException {
		System.out.println("收到客户端消息：" + message + "** ToId:" + id);
		try {
			if (!WebSocketServer.sessionPools.isEmpty()) {
				final WebSocketMsgVO messageVO = HnJsonUtils.jsonToBean(message, WebSocketMsgVO.class);
				// if (HnStringUtils.isBlank(messageVO.getCreateDateTime())) {
				// messageVO.setCreateDateTime(HnDateUtils.format(new Date(), HnDateUtils.HN_DATE_TIME_PATTERN));
				// }
				if (HnStringUtils.equals(messageVO.getSendMsgType(), WebSocketMsgVO.HeartBeat)
						&& HnStringUtils.isBlank(messageVO.getToId())) {
					messageVO.setToId(id);
				}

				for (final Map.Entry<String, Session> e : WebSocketServer.sessionPools.entrySet()) {
					final String key = e.getKey();
					if (key.indexOf(messageVO.getToId() + "_") == 0) {
						this.sendMessage(WebSocketServer.sessionPools.get(key), HnJsonUtils.jsonToString(messageVO));
					}
				}
			}
		} catch (final Exception e) {
			e.printStackTrace();
		}
	}

	// 错误时调用
	@OnError
	public void onError(final Session session, final Throwable throwable) {
		System.out.println("发生错误");
		throwable.printStackTrace();
	}

	// 发送消息
	private void sendMessage(final Session session, final String message) throws IOException {
		if (session != null) {
			synchronized (session) {
				System.out.println("发送数据：" + message);
				session.getBasicRemote().sendText(message);
			}
		}
	}

	public static void addOnlineCount() {
		WebSocketServer.onlineNum.incrementAndGet();
	}

	public static void subOnlineCount() {
		WebSocketServer.onlineNum.decrementAndGet();
	}
}
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

上面的代码里改变之处就是使用了sessionId，当发送消息时再做个循环处理即可，代码很简单，与前面的代码做个对比就清楚了。

PS:
1.前面的代码在大多数项目中基本已经够用了，但如果在生产环境下需要做分布式处理时还是会有问题的，因为由于后端代码放在不同服务器上，当前端每一次发送消息，进行连接时，都可能会连接到不同的服务器上，这就会导致发送的信息可能收不到，关于这种解决的方法也很多，例如使用redis的消息监听器，接下来有时间说说这块。

2.在正式部署时，可能是独立的部署在tomcat下，要先将WebSocketConfig文件注释掉，否则启动时会报错。