websocket实现客户端-服务端通信

迈不过友情╰ 2023-10-04 19:49 68阅读 0赞

实现websocket有三种方式

*  使用Java提供的@ServerEndpoint注解实现
 *  使用Spring提供的低层级WebSocket API实现
 *  使用STOMP消息实现

https://www.zifangsky.cn/1355.html 具体可看改文章详解

以下为第一种方式代码，先添加依赖

<dependency>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-websocket</artifactId>
                </dependency>

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.stereotype.Service;
    import org.springframework.web.socket.config.annotation.EnableWebSocket;
    import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    
    /**
     * websocket配置类
     * @author daiyg
     * @date 2021/8/24 17:19
     */
    @Service
    @Configuration
    @EnableWebSocket //开启对WebSocket的支持功能
    public class WebSocketConfig {
    
        @Bean
        public ServerEndpointExporter serverEndpointExporter(){
            return new ServerEndpointExporter();
        }
    
    }

逻辑实现类

import com.haileer.dd.dingdingserver.controller.TahDeviceController;
    import com.haileer.dd.dingdingserver.entity.vo.DeviceDto;
    import org.apache.logging.log4j.LogManager;
    import org.apache.logging.log4j.Logger;
    import org.springframework.beans.factory.annotation.Autowired;
    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.concurrent.ConcurrentHashMap;
    
    /**
     * @author daiyg
     * @date 2021/8/24 18:15
     */
    @Component
    @ServerEndpoint(value = "/websocket/{deviceNo}", subprotocols = {"protocol"})
    public class WebSocketServerSample {
    
        private static TahDeviceController tahDeviceController;
    
        private static int onlineCount = 0;
        private static ConcurrentHashMap<String, WebSocketServerSample> webSocketSet = new ConcurrentHashMap<>();
        private static  ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
    
        //与某个客户端的连接会话，需要通过它来给客户端发送数据
        private Session session;
        private static Logger log = LogManager.getLogger(WebSocketServerSample.class);
        private String id = "";
    
    
        /**
         * @Component下 @Autowired直接使用并不生效
         * 因为在使用@Component注解将bean实例化到spring容器内的时候，@Autowired是在这个bean之中的，@Autowired还未完成自动装载，所以导致service为null
         * 而放在方法上会在类加载后自动注入这个方法的参数，并执行一遍方法。
         */
        @Autowired
        public void setEmployeeServer(TahDeviceController tahDeviceController) {
            WebSocketServerSample.tahDeviceController = tahDeviceController;
        }
    
        /**
         * 连接建立成功调用的方法
         */
        @OnOpen
        public void onOpen(@PathParam(value = "deviceNo") String deviceNo, Session session) {
            this.session = session;
            this.id = deviceNo;//接收到发送消息的人员编号
            if(map.get(deviceNo)==null){
                map.put(deviceNo,deviceNo);
                webSocketSet.put(deviceNo, this);     //加入set中
                addOnlineCount();           //在线数加1
    
                //修改设备状态
                DeviceDto deviceDto = new DeviceDto();
                deviceDto.setDeviceNo(deviceNo);
                deviceDto.setStatus("1");
                tahDeviceController.updateStatus(deviceDto);
            }
            log.info("用户" + deviceNo + "连接成功！当前在线人数为" + getOnlineCount());
            try {
                sendMessage("连接成功");
            } catch (IOException e) {
                log.error("websocket IO异常");
            }
    
        }
    
        /**
         * 连接关闭调用的方法
         */
        @OnClose
        public void onClose() {
            if(map.get(this.id)!=null){
                map.remove(this.id);
                webSocketSet.remove(this.id);  //从set中删除
                subOnlineCount();           //在线数减1
                //修改设备状态
                DeviceDto deviceDto = new DeviceDto();
                deviceDto.setDeviceNo(id);
                deviceDto.setStatus("1");
                tahDeviceController.updateStatus(deviceDto);
            }
            log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
        }
    
        /**
         * 收到客户端消息后调用的方法
         *
         * @param message 客户端发送过来的消息
         */
        @OnMessage
        public void onMessage(String message) {
            log.info("来自"+this.id+"客户端的消息:" + message);
        }
    
        /**
         * @param session
         * @param error
         */
        @OnError
        public void onError(Session session, Throwable error) {
            log.error("发生错误");
            try {
                session.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            error.printStackTrace();
    
        }
    
        /**
         * 服务端主动推送消息
         *
         * @param message
         * @throws IOException
         */
        public void sendMessage(String message) throws IOException {
            this.session.getBasicRemote().sendText(message);
        }
    
        /**
         * 发送信息给指定ID用户，如果用户不在线则返回不在线信息给自己
         *
         * @param message
         * @param personKeys
         * @throws IOException
         */
        public static void sendtoUser(String message, String[] personKeys) throws IOException {
            //获取人员集合
            for (String personKey : personKeys) {
                //获取map中对应的链接
                if (map.get(personKey) != null) {
                    webSocketSet.get(personKey).sendMessage(message);
                } else {
                    //如果用户不在线则返回不在线信息给自己
                    log.error("当前用户不在线：" + personKey);
                }
            }
    
        }
    
    
        /**
         * 发送信息给所有人
         *
         * @param
         * @throws IOException
         */
        public void sendtoAll(String message) throws IOException {
            for (String key : webSocketSet.keySet()) {
                try {
                    webSocketSet.get(key).sendMessage(message);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
    
        public static synchronized int getOnlineCount() {
            return onlineCount;
        }
    
        public static synchronized void addOnlineCount() {
            WebSocketServerSample.onlineCount++;
        }
    
        public static synchronized void subOnlineCount() {
            WebSocketServerSample.onlineCount--;
        }
    
    }