引入pom依赖

<!--websocket连接需要使用到的包-->
<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.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;


/**
 * @author: tanghaizhi
 * @CreateTime: 2022/10/13 9:20
 * @Description: websocket配置,用于开启websocket支持
 */
@Configuration
@EnableWebSocket
public class WebSocketConfig {
          
   

    @Bean
    public ServerEndpointExporter serverEndpointExporter(){
          
   
        return new ServerEndpointExporter();
    }

}

WebSockt消息处理及推送

import lombok.Data;
import org.springframework.stereotype.Component;

import javax.websocket.OnOpen;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import javax.websocket.*;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author: tanghaizhi
 * @CreateTime: 2022/10/13 9:30
 * @Description:
 */
@Component
@Data
@ServerEndpoint(value = "/websocket/{client}}")
public class WebSocket {
          
   


    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;
    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    //也可以使用线程安全的map,在key中存放客户端连接的id等表唯一的东西，在发消息时可以精确取到和那个客户端的连接，不用群发消息
    private static CopyOnWriteArraySet<WebSocket> webSocketSet = new CopyOnWriteArraySet<WebSocket>();

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;



    /**
     * 建立连接时调用
     * @param session
     * @param config
     */
    @OnOpen
    public void onOpen(Session session, EndpointConfig config){
          
   
        System.out.println(session + "建立了连接");
        System.out.println(session.getRequestURI());
        this.session = session;
        //加入set中
        webSocketSet.add(this);
    }


    /**
     * 断开连接时调用
     * @param session
     */
    @OnClose
    public void onClose(Session session) {
          
   
        System.out.println(session + "断开了连接");
        webSocketSet.remove(this);
    }


    /**
     * 消息到达时调用
     *
     * @param
     * @param message
     * @param session
     */
    @OnMessage
    public String onMessage(String message, Session session){
          
   
        System.out.println(message);
//        try {
          
   
//            session.getBasicRemote().sendText(message);
//        } catch (IOException e) {
          
   
//            e.printStackTrace();
//        }
        return "{"taskId":"d4as5d4a56sd4","data":"码流文件","type":"3"}";
    }



    /**
     * 发生错误时调用
     *
     * @param session
     * @param throwable
     */
    @OnError
    public void onError(Session session, Throwable throwable) {
          
   
        System.out.println("发生错误！");
    }


    public void sendMessage(String message) throws IOException {
          
   
        this.session.getBasicRemote().sendText(message);
    }


    /**
     * 群发消息
     * 想主动给客服端推送消息时直接调用这个静态方法
     */
    public static void sendInfo(String message) throws IOException {
          
   
        for (WebSocket item : webSocketSet) {
          
   
            try {
          
   
                item.sendMessage(message);
            } catch (IOException e) {
          
   
                continue;
            }
        }
    }

}

测试