springBoot整合WebSocket实例（TCP、UDP）

墨蓝 2023-10-12 08:48 87阅读 0赞

## 一、Tcp api ##

### **1. 引入websocket依赖** ###

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

### **2. websocket配置类** ###

/**
     * WebSocket配置
     *
     * @author 
     * @version v1.0.0
     */
    @Configuration
    public class WebSocketConfig {
     
        @Bean
        public ServerEndpointExporter serverEndpointExporter() {
            return new ServerEndpointExporter();
        }
     
    }

### **3. websocket操作类** ###

/**
     * websocket server服务
     *
     * @author 
     * @version v1.0.0
     */
    @ServerEndpoint("/{userId}")
    @Component
    public class WebSocketServer {
        private static ApplicationContext applicationContext;
    
        public static void setApplicationContext(ApplicationContext applicationContext) {
            WebSocketServer.applicationContext = applicationContext;
        }
    
        private static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap<>();   // session管理map
        private Session session;    // 会话session
        private String userId="";   // 当前用户
    
        List dataList = new ArrayList();
    
        private static IotSensorInfoService iotSensorInfoService;
    
        private static BusinessStationService businessStationService;
    
        private static MqttPushClient mqttPushClient;
    
        @Autowired
        public void setBean(IotSensorInfoService iotSensorInfoService,BusinessStationService businessStationService,MqttPushClient mqttPushClient) {
            WebSocketServer.iotSensorInfoService = iotSensorInfoService;
            WebSocketServer.businessStationService = businessStationService;
            WebSocketServer.mqttPushClient = mqttPushClient;
        }
    
    
        /**
         * 初始化连接
         *
         * @param session 会话session
         * @param userId 当前用户id
         */
        @OnOpen
        public void onOpen(Session session, @PathParam("userId") String userId) {
            this.session = session;
            this.userId=userId;
     
            // 保存各用户的会话session
            if(webSocketMap.containsKey(userId)){
                webSocketMap.remove(userId);
                webSocketMap.put(userId, this);
            }else{
                webSocketMap.put(userId, this);
                //addOnlineCount();
            }
     
            try {
                sendMessage("connect success");
            } catch (IOException e) {
         
            }
        }
     
        /**
         * 连接关闭调用的方法
         */
        @OnClose
        public void onClose() {
            if(webSocketMap.containsKey(userId)){
                webSocketMap.remove(userId);
                //batchInsert();
            }
        }
     
        /**
         * 收到客户端消息后调用的方法
         *
         * @param message 客户端发送过来的消息
         */
        @OnMessage
        public void onMessage(String message, Session session) throws InterruptedException {
            RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
    
            try{
                Thread.sleep(50);
                // 消息放入线程池中处理（注：此处为我的业务逻辑处理，大家换成自己的即可）
                ThreadPoolUtil.execute(new MessageHandler(dataList , message, rabbitTemplate,iotSensorInfoService,businessStationService,mqttPushClient));
            }catch(Exception e){
                e.printStackTrace();
            }
        }
     
        /**
         * 异常处理
         * @param session
         * @param error
         */
        @OnError
        public void onError(Session session, Throwable error) {
            //batchInsert();
            error.printStackTrace();
        }
     
        public void sendMessage(String message) throws IOException {
            this.session.getBasicRemote().sendText(message);
        }
     
        /**
         * 发送自定义消息
         *
         * @param message 消息体
         * */
        public void sendInfo(String message) throws IOException {
     
            List<String> keys = new ArrayList<>(webSocketMap.keySet());
            if (CollectionUtils.isNotEmpty(keys)) {
                for (String key : keys) {
                    webSocketMap.get(key).sendMessage(message);
                }
            }
        }
    
        //分批入库（其他逻辑，可忽略）
        public void batchInsert(){
            int partialLimit = 100;
            //判断是否有必要分批
            if (partialLimit < dataList.size()) {
                //当前数据按限制条数可分为多少批次
                int part = dataList.size()/partialLimit;
                List<Integer> partList;
                for (int i = 0; i < part; i++) {
                    // 截取批次长度的list
                    partList = dataList.subList(0, partialLimit);
                    // 分批业务逻辑处理- 打印替代
                    System.out.println(partList);
                    // 去除已经处理的部分 (Arrays.asList()方式生成的数据不能进行此修改操作，会报错)
                    partList.clear();
                }
                // 获取最后一次截取后的剩余列表数据
                if (!dataList.isEmpty()) {
                    // 业务逻辑数据处理， - 打印替代
                    System.out.println(dataList);
                }
            } else {
                //直接入库
            }
        }
    }

**重要类方法说明:**  
(1)@ServerEndpoint(“/\{userId\}”)： 前端通过此 URI 和后端交互，建立连接，路径可自定义  
(2)@Component ：将此类交给 spring 管理  
(3)@OnOpen： websocket 建立连接的注解，前端触发上面 URI 时会进入此注解标注的方法  
(4)@OnMessage： 收到前端传来的消息后执行的方法，将消息数据处理逻辑写到此处  
(5)@OnClose： 顾名思义关闭连接，销毁 session

### **4. 测试** ###

（1）启动你的项目

（2）我们可以直接使用websocket在线测试工具进行测试，地址：

[websocket在线调试工具][websocket]

![051d0ab5278641ad9a803e89fd6402a3.png][]

（3）点击开启连接

（4）修改发送的内容，点击开始发送

（5）后端收到数据：

![3d484b7a4e1e4abca6226b7ef39e6fce.png][]

## 二、UDP api ##

### 1. 服务器 ###

package com.zlhy.websocket.controller;
    
    import com.github.pagehelper.util.StringUtil;
    import com.zlhy.websocket.util.ThreadPoolUtil;
    import com.zlhy.websocket.util.handler.MessageHandler;
    import org.springframework.beans.factory.annotation.Autowired;
    
    import java.io.IOException;
    import java.net.DatagramPacket;
    import java.net.DatagramSocket;
    import java.net.SocketException;
    
    
    // 服务器
    public class UdpTraServer {
        // 创建一个Socket 对象
        DatagramSocket socket = null;
    
        // 给服务器指定一个端口号
        public UdpTraServer(int port) throws SocketException {
            this.socket = new DatagramSocket(port);
        }
    
        public void start(WebSocketServer webSocketServer) throws IOException {
            System.out.println("数据接收服务器已启动!");
            while (true) {
                try {
                    // 1. 接收来自客户端的请求
                    DatagramPacket requestPacket = new DatagramPacket(new byte[1024], 1024);
                    if (requestPacket != null && requestPacket.getLength() != 0) {
                        socket.receive(requestPacket);
                        // 将请求转变为String类型能够更方便的去处理.
                        String request = new String(requestPacket.getData(), 0, requestPacket.getLength());
                        if (StringUtil.isNotEmpty(request)) {
                            // 2. 处理请求
                            //String response = process(request);
                            String response = "接收成功...";
                            // 3. 将请求打包成一个数据报
                            DatagramPacket responsePacket = new DatagramPacket(response.getBytes(), response.getBytes().length,
                                    requestPacket.getSocketAddress()); // 这里要指定客户端的IP地址和端口号
                            // 4. 发送给客户端
                            socket.send(responsePacket);
    
                            ThreadPoolUtil.execute(new MessageHandler(request, webSocketServer));
                        }
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    
    
        public static void main(String[] args) throws IOException {
            // 给服务器指定一个17173的端口号.
            UdpTraServer udpTraServer = new UdpTraServer(17173);
            udpTraServer.start();
        }
    
    }

### 2. 客户端 ###

package com.zlhy.websocket.controller;
    
    import java.io.IOException;
    import java.net.*;
    import java.util.Scanner;
     
    // UDP版本的回响客户端
    class UdpEchoClient {
     
        // 和客户端的一样，也是要通过这个类实例出来的对象来操作网卡
        private DatagramSocket clientSocket = null;
        // 同时还需要设定服务器的IP和端口，这样才能把消息发给客户端
        // 服务器则不用设定，因为服务器作为被动的接受数据的一方，这数据中就包含了客户端的IP和端口
        private String serverIp = null;
        private int serverPort = -1;
     
        // 构造方法
        // 在构造方法中设定客户端端口、服务器IP和服务器端口
        public UdpEchoClient(String serverIp, int serverPort) throws SocketException {
            // 建议随机分配客户端的端口
            // 因为随机分配的端口是空闲的端口，如果自己指定，则不一定是
            this.clientSocket = new DatagramSocket();
            this.serverIp = serverIp;
            this.serverPort = serverPort;
        }
     
        // 通过start方法启动客户端
        public void start() throws IOException {
            System.out.println("客户端启动！");
            while (true) {
                System.out.println("请输入：");
                Scanner scanner = new Scanner(System.in);
     
                // 输入数据
                String data = scanner.next();
                if (data.equals("exit")) {
                    clientSocket.close();
                    break;
                }
     
                // 收到数据后打包数据  同样是以DatagramPacket为单位
                // 这里把字符串转成字节，还有字节的实际个数也要传过去
                // InetAddress这个类中的getByName方法可以把"127.0.0.1"这个字符串转成32位二进制数
                DatagramPacket clientPacket = new DatagramPacket(data.getBytes(), data.getBytes().length,
                        InetAddress.getByName(serverIp), serverPort);
     
                // 发送数据
                clientSocket.send(clientPacket);
     
                // ... ... 等待服务器处理，然后接收处理后的数据
     
                // 接收数据
                DatagramPacket backedData = new DatagramPacket(new byte[8192], 8000);
                // 把收到的数据填到数组当中
                clientSocket.receive(backedData);
     
                // 解析数据 然后使用数据
                work(backedData);
            }
     
        }
     
        // 处理返回数据的方法
        private void work(DatagramPacket backedData) {
            String useData = new String(backedData.getData(), 0, backedData.getLength());
            // 解析好之后使用数据
            System.out.println(useData);
        }
    }
     
    public class UdpClient {
     
        public static void main(String[] args) throws IOException {
            UdpEchoClient client = new UdpEchoClient("127.0.0.1", 17173);
            client.start();
        }
     
    }

### 3. 测试 ###

![21ef7d0745c740d6b4ec0c6b5b896c02.png][]

[websocket]: http://wstool.jackxiang.com/
[051d0ab5278641ad9a803e89fd6402a3.png]: https://img-blog.csdnimg.cn/051d0ab5278641ad9a803e89fd6402a3.png
[3d484b7a4e1e4abca6226b7ef39e6fce.png]: https://img-blog.csdnimg.cn/3d484b7a4e1e4abca6226b7ef39e6fce.png
[21ef7d0745c740d6b4ec0c6b5b896c02.png]: https://img-blog.csdnimg.cn/21ef7d0745c740d6b4ec0c6b5b896c02.png