1. Websocket的优势:
那图说话:
区别是http协议只能通过客户端触发,服务端响应
WebSocket 特性:长连接, 可以弥补: 服务端主动推动消息到客户端。
2. pom需要增加:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency>
3. 开启websocket支持
@Configuration
public class WebSocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter() {
return new ServerEndpointExporter();
}
}
4. 实现webscocket的功能区:
@ServerEndpoint("/websocket/{token}") @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 userName = null; private static TokenService tokenService; public static void setTokenService(TokenService tokenService) { WebSocketServer.tokenService = tokenService; } /** * 连接建立成功调用的方法*/ @OnOpen public void onOpen(Session session,@PathParam("token") String token) { LoginUser loginUser = tokenService.getUserInfoByToken(token); SysUser user = loginUser.getUser(); String userName = user.getUserName(); this.userName = userName; this.session = session; if(webSocketMap.containsKey(userName)){ webSocketMap.remove(userName); webSocketMap.put(userName,this); //加入set中 }else{ webSocketMap.put(userName,this); //加入set中 addOnlineCount(); //在线数加1 } log.info("用户连接:"+userName+",当前在线人数为:" + getOnlineCount()); try { sendMessage("连接成功"); } catch (IOException e) { log.error("用户:"+userName+",网络异常!!!!!!"); } } /** * 连接关闭调用的方法 */ @OnClose public void onClose() { if(webSocketMap.containsKey(userName)){ webSocketMap.remove(userName); //从set中删除 subOnlineCount(); } log.info("用户退出:"+userName+",当前在线人数为:" + getOnlineCount()); } /** * 收到客户端消息后调用的方法 * * @param message 客户端发送过来的消息*/ @OnMessage public void onMessage(String message, Session session) throws IOException { log.info("用户消息:"+userName+",报文:"+ message); handleComingMsg(message); } private void handleComingMsg(String message) { //可以群发消息 //消息保存到数据库、redis if(StringUtils.isNotBlank(message)){ try { //解析发送的报文 JSONObject jsonObject = JSON.parseObject(message); //追加发送人(防止串改) jsonObject.put("fromUserId",this.userName); 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 } webSocketMap.get(this.userName).sendMessage(message); }catch (Exception e){ e.printStackTrace(); } } } /** * * @param session * @param error */ @OnError public void onError(Session session, Throwable error) { log.error("用户错误:"+this.userName+",原因:"+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) { log.info("发送消息到:"+userId+",报文:"+message); if(StringUtils.isNotBlank(userId)&&webSocketMap.containsKey(userId)){ try { webSocketMap.get(userId).sendMessage(message); } catch (IOException e) { log.error("[SOCKETSERVICE-sendInfo]: response:{}", e.getStackTrace()); } }else{ log.error("用户"+userId+",不在线!"); } } /** * 发送自定义消息 * */ public static void pushMsgForUser(String message, String userName){ log.info("发送消息到:"+userName+",报文:"+message); if(StringUtils.isNotBlank(userName)&&webSocketMap.containsKey(userName)){ try { webSocketMap.get(userName).sendMessage(message); } catch (IOException e) { log.error("[SOCKETSERVICE-pushMsgForUser]: response:{}", e.getStackTrace()); } }else{ log.error("用户"+userName+",不在线!"); } } public static synchronized int getOnlineCount() { return onlineCount; } public static synchronized void addOnlineCount() { WebSocketServer.onlineCount++; } public static synchronized void subOnlineCount() { WebSocketServer.onlineCount--; } }
5. 因为WebSocketServer类不能通过注入@Autowired.
所以暂时的解决思路是:
@SpringBootApplication public class WebSocketApplication { public static void main(String[] args) { ApplicationContext applicationContext = SpringApplication.run(JqylApplication.class, args); // 获取Spring IOC容器中的Service并注入 TokenService tokenService = applicationContext.getBean(TokenService.class); WebSocketServer.setTokenService(tokenService); } }