SpringWebSocketConfig配置

package com.meeno.chemical.socket.task.config;

import com.meeno.chemical.socket.task.handler.TaskProgressWebSocketHandler;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;

/**
 * @description: SpringBoot WebSocket config
 * @author: Wzq
 * @create: 2020-07-16 10:50
 */
@Configuration
@EnableWebSocket
public class SpringWebSocketConfig implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        // 注冊(cè)自定義消息處理虐译，消息路徑為`/ws/taskProgress`
        registry.addHandler(new TaskProgressWebSocketHandler(),"/ws/taskProgress").setAllowedOrigins("*");
    }
}

TaskProgressWebSocketHandler

package com.meeno.chemical.socket.task.handler;

import org.springframework.http.HttpHeaders;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.adapter.standard.StandardWebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @description: 任務(wù)進(jìn)度WebSocketHandler
 * @author: Wzq
 * @create: 2020-07-16 10:51
 */
public class TaskProgressWebSocketHandler extends TextWebSocketHandler {

    /**
     *  concurrent包的線程安全Map，用來(lái)存放每個(gè)客戶端對(duì)應(yīng)的MyWebSocket對(duì)象。
     */
    public final static ConcurrentHashMap<String, WebSocketSession> webSocketMap = new ConcurrentHashMap<String, WebSocketSession>();

    /**
     * 處理消息
     * @param session
     * @param message
     * @throws Exception
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        //在這里自定義消息處理...
        TextMessage textMessage = new TextMessage("${message body}");
        session.sendMessage(textMessage);
    }

    /**
     * 連接建立后
     * @param session
     * @throws Exception
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        StandardWebSocketSession standardWebSocketSession = (StandardWebSocketSession) session;
        List<String> taskNameList = standardWebSocketSession.getNativeSession().getRequestParameterMap().get("taskName");
        String taskName = taskNameList.get(0);
        //保存所有會(huì)話
        webSocketMap.put(taskName,session);
    }

    /**
     * 連接關(guān)閉后
     * @param session
     * @param status
     * @throws Exception
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {

        if(!webSocketMap.isEmpty()){
            // remove session after closed
            WebSocketSession webSocketSession = webSocketMap.get(session.getId());
            if(webSocketSession != null){
                webSocketMap.remove(session.getId());
            }
        }
    }


    /**
     * 發(fā)送消息給所有人
     * @param content
     */
    public static void sendMessageAll(String content){
        webSocketMap.forEach((taskNameKey,session) ->{
            TextMessage textMessage = new TextMessage(content);
            try {
                session.sendMessage(textMessage);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * 發(fā)送消息給指定某個(gè)任務(wù)
     * @param taskName
     * @param content
     */
    public static void sendMessage(String taskName,String content){
        webSocketMap.forEach((taskNameKey,session) ->{
            if(taskName.equals(taskNameKey)){
                TextMessage textMessage = new TextMessage(content);
                try {
                    session.sendMessage(textMessage);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

}

問(wèn)題

啟用是會(huì)和定時(shí)任務(wù)沖突
需要配置
ScheduledConfig.java

package com.meeno.chemical.common.scheduling.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

/**
 * @description: 任務(wù)config
 * @author: Wzq
 * @create: 2020-07-16 11:05
 */
@Configuration
public class ScheduledConfig {

    @Bean
    public TaskScheduler taskScheduler(){
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setPoolSize(10);
        taskScheduler.initialize();
        return taskScheduler;
    }

}