SpringBoot2.0集成WebSocket实现后台向前端推送信息

WebSocketServer

这就是重点了，核心都在这里。

• 因为WebSocket是类似客户端服务端的形式(采用ws协议)，那么这里的WebSocketServer其实就相当于一个ws协议的Controller
• 直接@ServerEndpoint("/imserver/{userId}") 、@Component启用即可，然后在里面实现@OnOpen开启连接，@onClose关闭连接，@onMessage接收消息等方法。
• 新建一个ConcurrentHashMap webSocketMap 用于接收当前userId的WebSocket，方便IM之间对userId进行推送消息。单机版实现到这里就可以。
• 集群版（多个ws节点）还需要借助mysql或者redis等进行处理，改造对应的sendMessage方法即可。

package com.softdev.system.demo.config;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;


/**
 * @author zhengkai.blog.csdn.net
 */
@ServerEndpoint("/imserver/{userId}")
@Component
public class WebSocketServer {

 static Log log=LogFactory.get(WebSocketServer.class);
 /**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/
 private static int onlineCount = 0;
 /**concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。*/
 private static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
 /**与某个客户端的连接会话，需要通过它来给客户端发送数据*/
 private Session session;
 /**接收userId*/
 private String userId="";

 /**
 * 连接建立成功调用的方法*/
 @OnOpen
 public void onOpen(Session session,@PathParam("userId") String userId) {
 this.session = session;
 this.userId=userId;
 if(webSocketMap.containsKey(userId)){
 webSocketMap.remove(userId);
 webSocketMap.put(userId,this);
 //加入set中
 }else{
 webSocketMap.put(userId,this);
 //加入set中
 addOnlineCount();
 //在线数加1
 }

 log.info("用户连接:"+userId+",当前在线人数为:" + getOnlineCount());

 try {
 sendMessage("连接成功");
 } catch (IOException e) {
 log.error("用户:"+userId+",网络异常!!!!!!");
 }
 }

 /**
 * 连接关闭调用的方法
 */
 @OnClose
 public void onClose() {
 if(webSocketMap.containsKey(userId)){
 webSocketMap.remove(userId);
 //从set中删除
 subOnlineCount();
 }
 log.info("用户退出:"+userId+",当前在线人数为:" + getOnlineCount());
 }

 /**
 * 收到客户端消息后调用的方法
 *
 * @param message 客户端发送过来的消息*/
 @OnMessage
 public void onMessage(String message, Session session) {
 log.info("用户消息:"+userId+",报文:"+message);
 //可以群发消息
 //消息保存到数据库、redis
 if(StringUtils.isNotBlank(message)){
 try {
 //解析发送的报文
 JSONObject jsonObject = JSON.parseObject(message);
 //追加发送人(防止串改)
 jsonObject.put("fromUserId",this.userId);
 String toUserId=jsonObject.getString("toUserId");
 //传送给对应toUserId用户的websocket
 if(StringUtils.isNotBlank(toUserId)&&webSocketMap.containsKey(toUserId)){
 webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());
 }else{
 log.error("请求的userId:"+toUserId+"不在该服务器上");
 //否则不在这个服务器上，发送到mysql或者redis
 }
 }catch (Exception e){
 e.printStackTrace();
 }
 }
 }

 /**
 *
 * @param session
 * @param error
 */
 @OnError
 public void onError(Session session, Throwable error) {
 log.error("用户错误:"+this.userId+",原因:"+error.getMessage());
 error.printStackTrace();
 }
 /**
 * 实现服务器主动推送
 */
 public void sendMessage(String message) throws IOException {
 this.session.getBasicRemote().sendText(message);
 }


 /**
 * 发送自定义消息
 * */
 public static void sendInfo(String message,@PathParam("userId") String userId) throws IOException {
 log.info("发送消息到:"+userId+"，报文:"+message);
 if(StringUtils.isNotBlank(userId)&&webSocketMap.containsKey(userId)){
 webSocketMap.get(userId).sendMessage(message);
 }else{
 log.error("用户"+userId+",不在线！");
 }
 }

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

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

 public static synchronized void subOnlineCount() {
 WebSocketServer.onlineCount--;
 }
}

消息推送

至于推送新信息，可以再自己的Controller写个方法调用WebSocketServer.sendInfo();即可

import com.softdev.system.demo.config.WebSocketServer;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.ModelAndView;
import java.io.IOException;

/**
 * WebSocketController
 * @author zhengkai.blog.csdn.net
 */
@RestController
public class DemoController {

 @GetMapping("index")
 public ResponseEntity<String> index(){
 return ResponseEntity.ok("请求成功");
 }

 @GetMapping("page")
 public ModelAndView page(){
 return new ModelAndView("websocket");
 }

 @RequestMapping("/push/{toUserId}")
 public ResponseEntity<String> pushToWeb(String message, @PathVariable String toUserId) throws IOException {
 WebSocketServer.sendInfo(message,toUserId);
 return ResponseEntity.ok("MSG SEND SUCCESS");
 }
}

页面发起

页面用js代码调用websocket，当然，太古老的浏览器是不行的，一般新的浏览器或者谷歌浏览器是没问题的。还有一点，记得协议是ws的，如果使用了一些路径类，可以replace(“http”,“ws”)来替换协议。

<!DOCTYPE html>
<html>
<head>
 <meta charset="utf-8">
 <title>websocket通讯</title>
</head>
<script src="/uploads/allimg/210325/1Q2133041-2.jpg"></script>
<script>
 var socket;
 function openSocket() {
 if(typeof(WebSocket) == "undefined") {
 console.log("您的浏览器不支持WebSocket");
 }else{
 console.log("您的浏览器支持WebSocket");
 //实现化WebSocket对象，指定要连接的服务器地址与端口 建立连接
 //等同于socket = new WebSocket("ws://localhost:8888/xxxx/im/25");
 //var socketUrl="${request.contextPath}/im/"+$("#userId").val();
 var socketUrl="http://blog.iis7.com/' defined in class path resource [com/xxx/WebSocketConfig.class]: Invocation of init method failed; nested exception is java.lang.IllegalStateException: javax.websocket.server.ServerContainer not available


感谢@来了老弟儿 的反馈：
如果tomcat部署一直报这个错，请移除 WebSocketConfig 中@Bean ServerEndpointExporter 的注入 。
ServerEndpointExporter 是由Spring官方提供的标准实现，用于扫描ServerEndpointConfig配置类和@ServerEndpoint注解实例。使用规则也很简单：
如果使用默认的嵌入式容器 比如Tomcat 则必须手工在上下文提供ServerEndpointExporter。如果使用外部容器部署war包，则不需要提供提供ServerEndpointExporter，因为此时SpringBoot默认将扫描服务端的行为交给外部容器处理，所以线上部署的时候要把WebSocketConfig中这段注入bean的代码注掉。 正式项目的前端WebSocket框架 GoEasy
感谢kkatrina的补充，正式的项目中，一般是用第三方websocket框架来做，稳定性、实时性有保证的多，也会包括一些心跳、重连机制。

GoEasy专注于服务器与浏览器,浏览器与浏览器之间消息推送,完美兼容世界上的绝大多数浏览器,包括IE6, IE7之类的非常古老的浏览器。支持Uniapp,各种小程序，react，vue等所有主流Web前端技术。

GoEasy采用 发布/订阅 的消息模式,帮助您非常轻松的实现一对一,一对多的通信。

https://www.goeasy.io/cn/doc/

@Component和@ServerEndpoint关于是否单例模式，能否使用static Map等一些问题的解答
看到大家都在热心的讨论关于是否单例模式这个问题，请大家相信自己的直接，如果websocket是单例模式，还怎么服务这么多session呢。

  
websocket是原型模式，@ServerEndpoint每次建立双向通信的时候都会创建一个实例，区别于spring的单例模式。Spring的@Component默认是单例模式，请注意，默认 而已，是可以被改变的。
  
这里的@Component仅仅为了支持@Autowired依赖注入使用，如果不加则不能注入任何东西，为了方便。
  
什么是prototype 原型模式？ 基本就是你需要从A的实例得到一份与A内容相同，但是又互不干扰的实例B的话，就需要使用原型模式。关于在原型模式下使用static 的webSocketMap，请注意这是ConcurrentHashMap ，也就是线程安全/线程同步的，而且已经是静态变量作为全局调用，这种情况下是ok的，或者大家如果有顾虑或者更好的想法的化，可以进行改进。
  
例如使用一个中间类来接收和存放session。为什么每次都@OnOpen都要检查webSocketMap.containsKey(userId) ，首先了为了代码强壮性考虑，假设代码以及机制没有问题，那么肯定这个逻辑是废的对吧。
  
但是实际使用的时候发现偶尔会出现重连失败或者其他原因导致之前的session还存在，这里就做了一个清除旧session，迎接新session的功能。

Vue版本的websocket连接
感谢**@GzrStudy**的贡献，供大家参考。

<script>
export default {
 data() {
 return {
 socket:null,
 userId:localStorage.getItem("ms_uuid"),
 toUserId:'2',
 content:'3'
 }
 },
 methods: {
 openSocket() {
 if (typeof WebSocket == "undefined") {
 console.log("您的浏览器不支持WebSocket");
 } else {
 console.log("您的浏览器支持WebSocket");
 //实现化WebSocket对象，指定要连接的服务器地址与端口 建立连接
 //等同于socket = new WebSocket("ws://localhost:8888/xxxx/im/25");
 //var socketUrl="${request.contextPath}/im/"+$("#userId").val();
 var socketUrl =
 "http://localhost:8081/imserver/" + this.userId;
 socketUrl = socketUrl.replace("https", "ws").replace("http", "ws");
 console.log(socketUrl);
 if (this.socket != null) {
 this.socket.close();
 this.socket = null;
 }
 this.socket = new WebSocket(socketUrl);
 //打开事件
 this.socket = new WebSocket(socketUrl);
 //打开事件
 this.socket.onopen = function() {
 console.log("websocket已打开");
 //socket.send("这是来自客户端的消息" + location.href + new Date());
 };
 //获得消息事件
 this.socket.onmessage = function(msg) {
 console.log(msg.data);
 //发现消息进入 开始处理前端触发逻辑
 };
 //关闭事件
 this.socket.onclose = function() {
 console.log("websocket已关闭");
 };
 //发生了错误事件
 this.socket.onerror = function() {
 console.log("websocket发生了错误");
 };
 }
 },
 sendMessage() {
 if (typeof WebSocket == "undefined") {
 console.log("您的浏览器不支持WebSocket");
 } else {
 console.log("您的浏览器支持WebSocket");
 console.log(
 '{"toUserId":"' +
 this.toUserId +
 '","contentText":"' +
 this.content +
 '"}'
 );
 this.socket.send(
 '{"toUserId":"' +
 this.toUserId +
 '","contentText":"' +
 this.content +
 '"}'
 );
 
 }
}