SpringBoot整合WebSocket

超、凢脫俗 2022-12-07 11:39 208阅读 0赞

项目demo源码 : 链接：https://pan.baidu.com/s/17vn27YKAX\_udvaA8QfJ0cw     提取码：zsq6   
这个是之前码云上面找的,demo测试可用,大佬位置忘记了,抱歉

### 核心代码1: ###

/**
     * WebScoket配置处理器
     */
    @Configuration
    public class WebSocketConfig {
        /**
         * ServerEndpointExporter 作用
         * <p>
         * 这个Bean会自动注册使用@ServerEndpoint注解声明的websocket endpoint
         *
         * @return
         */
        @Bean
        public ServerEndpointExporter serverEndpointExporter() {
            return new ServerEndpointExporter();
        }
    
    }

### 核心代码2: ###

@ServerEndpoint("/webSocket/{username}")
    @Component
    public class WebSocketServer {
    	 //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
        private static AtomicInteger onlineNum = new AtomicInteger();
    
        //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。
        private static ConcurrentHashMap<String, Session> sessionPools = new ConcurrentHashMap<>();
    
        //发送消息 5555
        public void sendMessage(Session session, String message) throws IOException {
            if(session != null){
                synchronized (session) {
                    System.out.println("发送数据：" + message);
                    session.getBasicRemote().sendText(message);
                }
            }
        }
        //给指定用户发送信息
        public void sendInfo(String userName, String message){
            Session session = sessionPools.get(userName);
            try {
                sendMessage(session, message);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        // 群发消息  4444(群发通知某某上线了)
        public void broadcast(String message){
        	for (Session session: sessionPools.values()) {
                try {
                    sendMessage(session, message);
                } catch(Exception e){
                    e.printStackTrace();
                    continue;
                }
            }
        }
    
        //建立连接成功调用  //33333
        @OnOpen
        public void onOpen(Session session, @PathParam(value = "username") String userName){
            sessionPools.put(userName, session);
            addOnlineCount();
            System.out.println(userName + "加入webSocket！当前人数为" + onlineNum);
            // 广播上线消息
            SocketMessage msg = new SocketMessage();
            msg.setDate(new Date());
            msg.setTo("0");
            msg.setText(userName);
            broadcast(JSON.toJSONString(msg,true));
        }
    
        //关闭连接时调用
        @OnClose
        public void onClose(@PathParam(value = "username") String userName){
            sessionPools.remove(userName);
            subOnlineCount();
            System.out.println(userName + "断开webSocket连接！当前人数为" + onlineNum);
            // 广播下线消息
            SocketMessage msg = new SocketMessage();
            msg.setDate(new Date());
            msg.setTo("-2");
            msg.setText(userName);
            broadcast(JSON.toJSONString(msg,true));
        }
    
        //收到客户端信息后，根据接收人的username把消息推下去或者群发
        // to=-1群发消息
        @OnMessage
        public void onMessage(String message) throws IOException{
            System.out.println("server get" + message);
            SocketMessage msg=JSON.parseObject(message, SocketMessage.class);
    		msg.setDate(new Date());
    		if (msg.getTo().equals("-1")) {
    			broadcast(JSON.toJSONString(msg,true));
    		} else {
    			sendInfo(msg.getTo(), JSON.toJSONString(msg,true));
    		}
        }
    
        //错误时调用
        @OnError
        public void onError(Session session, Throwable throwable){
            System.out.println("发生错误");
            throwable.printStackTrace();
        }
    
        public static void addOnlineCount(){
            onlineNum.incrementAndGet();
        }
    
        public static void subOnlineCount() {
            onlineNum.decrementAndGet();
        }
        
        public static AtomicInteger getOnlineNumber() {
            return onlineNum;
        }
        
        public static ConcurrentHashMap<String, Session> getSessionPools() {
            return sessionPools;
        }
    }