redis订阅发布实现websocket集群.zip

package com.example.demo.websocket; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.example.demo.entity.MallAd; import com.example.demo.model.WebSocketClient; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.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.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; /** * websocket连接处理类 */ @ServerEndpoint(value = "/websocket/{userName}") @Component public class WebSocketService { private static final Logger log = LoggerFactory.getLogger(WebSocketService.class); //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。 private static int onlineCount = 0; //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。 private static ConcurrentHashMap<String, WebSocketClient> webSocketMap = new ConcurrentHashMap<>(); /** * 与某个客户端的连接会话，需要通过它来给客户端发送数据 */ private Session session; /** * 接收userName */ private String userName = ""; /** * 连接建立成功调用的方法 */ @OnOpen public void onOpen(Session session, @PathParam("userName") String userName) { if (!webSocketMap.containsKey(userName)) { addOnlineCount(); // 在线数 +1 } this.session = session; this.userName = userName; WebSocketClient client = new WebSocketClient(); client.setSession(session); client.setUri(session.getRequestURI().toString()); webSocketMap.put(userName, client); log.info("----------------------------------------------------------------------------"); log.info("用户连接:" + userName + ",当前在线人数为:" + getOnlineCount()); try { sendMessage("来自后台的反馈：连接成功"); } catch (IOException e) { log.error("用户:" + userName + ",网络异常!!!!!!"); } } /** * 连接关闭调用的方法 */ @OnClose public void onClose() { if (webSocketMap.containsKey(userName)) { webSocketMap.remove(userName); if (webSocketMap.size() > 0) { //从set中删除 subOnlineCount(); } } log.info("----------------------------------------------------------------------------"); log.info(userName + "用户退出,当前在线人数为:" + getOnlineCount()); } /** * 收到客户端消息后调用的方法 * * @param message 客户端发送过来的消息 */ @OnMessage public void onMessage(String message, Session session) { try { JSONObject jsonObject = JSONObject.parseObject(message); String type = jsonObject.getString("type"); log.info("收到用户消息:" + userName + ",报文:" + message); if (type.equals("ping")) { //心跳 Map<String, Object> params = new HashMap<String, Object>(); params.put("type", "pong"); session.getBasicRemote().sendText(JSON.toJSONString(params)); } } catch (IOException e) { e.printStackTrace(); } //可以群发消息 //消息保存到数据库、redis if (StringUtils.isNotBlank(message)) { } } /** * 发生错误 * * @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 { synchronized (session) { this.session.getBasicRemote().sendText(message); } } /** * 向指定客户端发送消息 * * @param userName * @param message */ public static void sendMessage(String userName, String message) { try { WebSocketClient webSocketClient = webSocketMap.get(userName); if (webSocketClient != null) { webSocketClient.getSession().getBasicRemote().sendText(message); } } catch (IOException e) { e.printStackTrace(); throw new RuntimeException(e.getMessage()); } } /** * 给所有在线的，发送消息 */ public static void sendAll(List<MallAd> list) { String text = JSONObject.toJSONString(list); webSocketMap.values().forEach(webSocket -> webSocket.getSession().getAsyncRemote().sendText(text)); } public static synchronized int getOnlineCount() { return onlineCount; } public static synchronized void addOnlineCount() { WebSocketService.onlineCount++; } public static synchronized void subOnlineCount() { WebSocketService.onlineCount--; } public static void setOnlineCount(int onlineCount) { WebSocketService.onlineCount = onlineCount; } public static ConcurrentHashMap<String, WebSocketClient> getWebSocketMap() { return webSocketMap; } public static void setWebSocketMap(ConcurrentHashMap<String, WebSocketClient> webSocketMap) { WebSocketService.webSocketMap = webSocketMap; } public Session getSession() { return session; } public void setSession(Session session) { this.session = session; } public String getUserName() { return userName; } public void setUserName(String userName) { this.userName = userName; } }