SpringBoot2.x集成WebSocket实现广播和点对点推送消息

?WebSocket是一种在单个TCP连接上进行全双工通信的协议。WebSocket 协议在2008年诞生,2011年成为国际标准。现在所有浏览器都已经支持了。它的最大特点就是,服务器可以主动向客户端推送信息,客户端也可以主动向服务器发送信息,是真正的双向平等对话,属于服务器推送技术的一种。
?最近有需求使用WebSocket作为中间件去转发数据,于是就使用SpringBoot2.x与其进行整合,在此期间遇到了一些问题,总算是都解决了。话不多说,直接上代码。
1、添加jar支持,build.gradle如下:

plugins { id 'org.springframework.boot' version '2.1.3.RELEASE' id 'java' id 'idea' // id 'war' }apply plugin: 'io.spring.dependency-management'group = 'com.tiantian' version = '1.0.0' sourceCompatibility = '1.8' targetCompatibility = '1.8'repositories { mavenCentral() }dependencies { testImplementation ( 'org.springframework.boot:spring-boot-starter-test' )implementation ( 'org.springframework.boot:spring-boot-starter-web', 'org.springframework.boot:spring-boot-starter-websocket', //'org.yeauty:netty-websocket-spring-boot-starter:0.7.4', 'org.springframework.boot:spring-boot-starter-thymeleaf', 'org.apache.commons:commons-lang3:3.8.1', 'com.alibaba:fastjson:1.2.56' )// lombok支持,打包时处理 compileOnly ('org.projectlombok:lombok:1.18.6') annotationProcessor 'org.projectlombok:lombok:1.18.6' }// 使用bootJar打jar包 jar { baseName = 'webserver' version = '1.0.0' manifest { attributes "Manifest-Version": 1.0, 'Main-Class': 'com.tiantian.webserver.WebserverApplication' } }// 解决代码中的中文编译报错 tasks.withType(JavaCompile) { options.encoding = "UTF-8" }

说明:此项目是打成jar运行的,当然也可以打成war放在Tomcat上运行。
2、配置开启WebSocket功能
/** * WebSocket配置 * * @author yueli.liao * @date 2019-03-12 11:25 */ @Configuration public class WebSocketConfig {/** * 开启WebSocket功能 * * @return */ @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); }}

3、WebSocket核心服务
/** * WebSocket服务入口 * * @author yueli.liao * @date 2019-03-12 15:16 */ @Slf4j @Component @ServerEndpoint("/websocket/{id}") public class WebSocketServer {// 客户端ID private String id = ""; // 与某个客户端的连接会话,需要通过它来给客户端发送数据 private Session session; // 记录当前在线连接数(为保证线程安全,须对使用此变量的方法加lock或synchronized) private static int onlineCount = 0; // 用来存储当前在线的客户端(此map线程安全) private static ConcurrentHashMap webSocketMap = new ConcurrentHashMap<>(); /** * 连接建立成功后调用 */ @OnOpen public void onOpen(@PathParam(value = "https://www.it610.com/article/id") String id, Session session) { this.session = session; this.id = id; // 接收到发送消息的客户端编号 webSocketMap.put(id, this); // 加入map中 addOnlineCount(); // 在线数加1 log.info("客户端" + id + "加入,当前在线数为:" + getOnlineCount()); try { sendMessage("WebSocket连接成功"); } catch (IOException e) { log.error("WebSocket IO异常"); } }/** * 连接关闭时调用 */ @OnClose public void onClose() { webSocketMap.remove(this); // 从map中删除 subOnlineCount(); // 在线数减1 log.info("有一连接关闭,当前在线数为:" + getOnlineCount()); }/** * 收到客户端消息后调用 * * @param message 客户端发送过来的消息
*消息格式:内容 - 表示群发,内容|X - 表示发给id为X的客户端 * @param session */ @OnMessage public void onMessage(String message, Session session) { log.info("来自客户端的消息:" + message); String[] messages = message.split("[|]"); try { if (messages.length > 1) { sendToUser(messages[0], messages[1]); } else { sendToAll(messages[0]); } } catch (IOException e) { log.error(e.getMessage(), e); } }/** * 发生错误时回调 * * @param session * @param error */ @OnError public void onError(Session session, Throwable error) { log.error("WebSocket发生错误"); error.printStackTrace(); }/** * 推送信息给指定ID客户端,如客户端不在线,则返回不在线信息给自己 * * @param message 客户端发来的消息 * @param sendClientId 客户端ID * @throws IOException */ public void sendToUser(String message, String sendClientId) throws IOException { if (webSocketMap.get(sendClientId) != null) { if (!id.equals(sendClientId)) { webSocketMap.get(sendClientId).sendMessage("客户端" + id + "发来消息:" + "
" + message); } else { webSocketMap.get(sendClientId).sendMessage(message); } } else { // 如客户端不在线,则返回不在线信息给自己 sendToUser("当前客户端不在线", id); } }/** * 推送发送信息给所有人 * * @param message 要推送的消息 * @throws IOException */ public void sendToAll(String message) throws IOException { for (String key : webSocketMap.keySet()) { webSocketMap.get(key).sendMessage(message); } }/** * 推送消息 * * @param message 要推送的消息 * @throws IOException */ private void sendMessage(String message) throws IOException { this.session.getBasicRemote().sendText(message); }private static synchronized int getOnlineCount() { return onlineCount; }private static synchronized void addOnlineCount() { WebSocketServer.onlineCount++; }private static synchronized void subOnlineCount() { WebSocketServer.onlineCount--; }}

4、为解决大文本(超过WebSocket协议限制的长度)不能发送的问题,有两种解决办法,一种是在Tomcat \conf\web.xml文件添加如下代码:
【SpringBoot2.x集成WebSocket实现广播和点对点推送消息】org.apache.tomcat.websocket.textBufferSize1024000 org.apache.tomcat.websocket.binaryBufferSize1024000

另一种是在服务启动时动态进行设置,在项目中添加如下代码:
/** * 解决WebSocket传输内容过长的问题 * 注:若发送内容过长时,服务端无内容输出,连接会自动关闭 * * @author yueli.liao * @date 2019-03-13 13:35 */ @Slf4j @Configuration @ComponentScan @EnableAutoConfiguration public class WebAppRootContext implements ServletContextInitializer {@Value("${websocket.bufferSize}") private String bufferSize; @Override public void onStartup(ServletContext servletContext) throws ServletException { log.info("WebSocket最大文本长度:" + bufferSize); servletContext.addListener(WebAppRootListener.class); servletContext.setInitParameter("org.apache.tomcat.websocket.textBufferSize", bufferSize); }}

代码中的websocket.bufferSize,在application.properties进行配置,如下:
#websocket textSize(单位为byte) websocket.bufferSize=1024000

说明:推荐使用第二种方式,这样可移植性强,对部署和开发都很方便。
5、测试,也有两种方式,采用哪种或两种都可以。
  • 一是在Google或Firefox浏览器中安装WebSocket Client插件进行测试;
  • 二是在项目中写相关的JS进行测试,代码如下:
WebSocket测试 - 锐客网 Welcome

说明:此webserver可以进行点对点推送消息,也可以进行广播。源代码已上传到GitHub,可自行下载。

    推荐阅读