当前位置：首页 > article >正文

webSocket发送实时通知实例

article 2025/2/22 6:03:02

在 Spring Boot 中实现前端调用接口执行任务并通过 WebSocket 实时推送任务状态，可以按照以下步骤操作：

1. 添加依赖

在 pom.xml 中添加 WebSocket 和 JSON 支持依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

2. 配置 WebSocket

创建一个 WebSocket 配置类，启用 STOMP 协议和消息代理：

import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // 前端连接端点
        registry.addEndpoint("/ws")
                .setAllowedOrigins("*") // 允许跨域
                .withSockJS();         // 支持 SockJS
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // 启用内存消息代理，前缀为 /topic 的消息会发送到代理
        registry.enableSimpleBroker("/topic");
        // 客户端发送消息的前缀（可选，若需双向通信）
        registry.setApplicationDestinationPrefixes("/app");
    }
}

3. 定义任务状态消息对象

创建一个 DTO 类表示任务状态消息：

public class TaskStatusMessage {
    private String taskId;
    private String status;  // "STARTED", "SUCCESS", "FAILED"
    private String message;

    // 构造方法、Getter 和 Setter
}

4. 编写任务服务与 WebSocket 推送

在服务层中执行任务并通过 SimpMessagingTemplate 推送消息：

import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

@Service
public class TaskService {

    private final SimpMessagingTemplate messagingTemplate;

    public TaskService(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    @Async // 异步执行任务
    public void executeTask(String taskId) {
        try {
            // 发送任务开始通知
            sendStatus(taskId, "STARTED", "任务开始执行");

            // 模拟耗时操作
            Thread.sleep(5000);

            // 模拟成功或失败
            boolean success = Math.random() > 0.5;
            if (success) {
                sendStatus(taskId, "SUCCESS", "任务执行成功");
            } else {
                sendStatus(taskId, "FAILED", "任务执行失败");
            }
        } catch (InterruptedException e) {
            sendStatus(taskId, "FAILED", "任务被中断");
        }
    }

    private void sendStatus(String taskId, String status, String message) {
        TaskStatusMessage msg = new TaskStatusMessage();
        msg.setTaskId(taskId);
        msg.setStatus(status);
        msg.setMessage(message);
        // 推送消息到 /topic/task-status
        messagingTemplate.convertAndSend("/topic/task-status", msg);
    }
}

5. 创建任务触发接口

编写 REST 接口触发任务并返回任务 ID：

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.UUID;

@RestController
public class TaskController {

    private final TaskService taskService;

    public TaskController(TaskService taskService) {
        this.taskService = taskService;
    }

    @GetMapping("/start-task")
    public String startTask() {
        String taskId = UUID.randomUUID().toString();
        taskService.executeTask(taskId);
        return taskId; // 返回任务ID供前端订阅状态
    }
}

6. 前端实现（JavaScript）

使用 SockJS 和 Stomp.js 连接 WebSocket 并订阅状态：

<!DOCTYPE html>
<html>
<head>
    <title>任务状态监控</title>
    <script src="https://cdn.jsdelivr.net/npm/sockjs-client@1.5.0/dist/sockjs.min.js"></script>
    <script src="https://cdn.jsdelivr.net/npm/@stomp/stompjs@6.1.2/dist/stomp.umd.min.js"></script>
</head>
<body>
    <button onclick="startTask()">启动任务</button>
    <div id="status"></div>

    <script>
        let stompClient = null;
        const taskIdElement = document.createElement('div');
        document.body.appendChild(taskIdElement);

        // 连接 WebSocket
        function connect() {
            const socket = new SockJS('http://localhost:8080/ws');
            stompClient = Stomp.over(socket);
            stompClient.connect({}, () => {
                console.log('WebSocket 已连接');
            });
        }

        // 启动任务
        function startTask() {
            fetch('/start-task')
                .then(response => response.text())
                .then(taskId => {
                    taskIdElement.textContent = `任务ID: ${taskId}`;
                    subscribeToTaskStatus(taskId);
                });
        }

        // 订阅任务状态
        function subscribeToTaskStatus(taskId) {
            if (stompClient) {
                stompClient.subscribe(`/topic/task-status`, (message) => {
                    const statusMsg = JSON.parse(message.body);
                    if (statusMsg.taskId === taskId) {
                        document.getElementById('status').innerHTML += 
                            `<p>状态: ${statusMsg.status} - ${statusMsg.message}</p>`;
                    }
                });
            }
        }

        // 初始化连接
        connect();
    </script>
</body>
</html>

7. 启用异步支持

在 Spring Boot 主类添加 @EnableAsync：

import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableAsync
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

8. 运行与测试

启动 Spring Boot 应用。
访问前端页面，点击按钮启动任务。
观察实时状态更新。

关键点说明

WebSocket 连接：前端通过 /ws 端点连接，使用 SockJS 和 STOMP。
消息推送：服务端通过 SimpMessagingTemplate 向 /topic/task-status 发送消息。
异步任务：使用 @Async 避免阻塞主线程，需配置线程池（默认使用 SimpleAsyncTaskExecutor）。
消息过滤：前端根据 taskId 过滤属于自己的任务状态。

扩展优化

线程池配置：自定义 @Async 的线程池，避免无限制创建线程。

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(50);
        executor.initialize();
        return executor;
    }
}