Spring Boot 集成 WebSocket 实现后台向前端推送信息

Spring Boot 集成 WebSocket 实现后台向前端推送信息 | 极客日志

<dependencies>
    <!-- Spring Boot WebSocket Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>
    <!-- Spring Boot Web Starter (通常包含在内，但显式添加也无妨) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- 可选：用于简化开发的热部署 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-devtools</artifactId>
        <scope>runtime</scope>
        <optional>true</optional>
    </dependency>
    <!-- 可选：JSON 处理 -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

implementation 'org.springframework.boot:spring-boot-starter-websocket'
implementation 'org.springframework.boot:spring-boot-starter-web'
runtimeOnly 'org.springframework.boot:spring-boot-devtools'

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

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

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

@ServerEndpoint("/ws/echo")
public class EchoWebSocket {
    // 存储所有连接的会话，用于广播
    private static final CopyOnWriteArraySet<Session> sessions = new CopyOnWriteArraySet<>();

    @OnOpen
    public void onOpen(Session session) {
        sessions.add(session);
        System.out.println("新连接加入，当前连接数：" + sessions.size());
        // 可以给客户端发送欢迎消息
        try {
            session.getBasicRemote().sendText("连接成功，欢迎！");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        System.out.println("收到消息：" + message);
        // 回声：将消息返回给发送者
        try {
            session.getBasicRemote().sendText("Echo: " + message);
        } catch (IOException e) {
            e.printStackTrace();
        }
        // 也可以广播给所有人（可选）
        broadcast("用户说：" + message);
    }

    @OnClose
    public void onClose(Session session) {
        sessions.remove(session);
        System.out.println("连接关闭，当前连接数：" + sessions.size());
    }

    @OnError
    public void onError(Session session, Throwable error) {
        error.printStackTrace();
    }

    // 广播消息给所有客户端
    public static void broadcast(String message) {
        for (Session session : sessions) {
            try {
                session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

<!DOCTYPE html>
<html>
<head>
    <title>WebSocket Echo Test</title>
</head>
<body>
    <input type="text" placeholder="输入消息" />
    <button onclick="sendMessage()">发送</button>
    <div id="messages"></div>
    <script>
        var ws = new WebSocket("ws://localhost:8080/ws/echo");
        ws.onopen = function() {
            appendMessage("连接已建立");
        };
        ws.onmessage = function(event) {
            appendMessage("收到：" + event.data);
        };
        ws.onclose = function() {
            appendMessage("连接关闭");
        };
        ws.onerror = function(error) {
            appendMessage("错误：" + error);
        };
        function sendMessage() {
            var input = document.getElementById("messageInput");
            var msg = input.value;
            ws.send(msg);
            appendMessage("发送：" + msg);
            input.value = '';
        }
        function appendMessage(msg) {
            var div = document.getElementById("messages");
            div.innerHTML += "<p>" + msg + "</p>";
        }
    </script>
</body>
</html>

import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
@EnableScheduling
public class PushTask {
    @Scheduled(fixedRate = 5000)
    public void pushTime() {
        String message = "当前时间：" + System.currentTimeMillis();
        EchoWebSocket.broadcast(message);
    }
}

COMMAND header1:value1 header2:value2 body

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 StompWebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // 注册一个 STOMP 端点，客户端通过这个端点进行连接
        registry.addEndpoint("/ws-stomp")
                .setAllowedOrigins("*") // 允许跨域
                .withSockJS(); // 启用 SockJS 支持，用于降级方案
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // 设置应用目的地前缀，即客户端发送消息的目的地前缀
        registry.setApplicationDestinationPrefixes("/app");
        // 启用简单消息代理，并设置消息代理前缀（订阅前缀）
        registry.enableSimpleBroker("/topic", "/queue");
        // 点对点使用的订阅前缀（默认是/user/）
        registry.setUserDestinationPrefix("/user");
    }
}

import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Controller;

@Controller
public class WebSocketController {
    /**
     * 客户端发送消息到 /app/hello，服务器处理后，将结果广播给所有订阅了 /topic/greetings 的客户端
     */
    @MessageMapping("/hello")
    @SendTo("/topic/greetings")
    public Greeting greeting(HelloMessage message) throws Exception {
        // 模拟处理延迟
        Thread.sleep(1000);
        return new Greeting("Hello, " + message.getName() + "!");
    }
}

// 消息体类
class HelloMessage {
    private String name;
    // getter/setter...
}

class Greeting {
    private String content;
    // constructor, getter/setter...
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.GetMapping;

@RestController
public class PushController {
    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @GetMapping("/push")
    public String pushToAll() {
        // 向所有订阅了 /topic/news 的客户端推送消息
        messagingTemplate.convertAndSend("/topic/news", "突发新闻：Spring Boot 3.0 发布！");
        return "推送成功";
    }

    @GetMapping("/push/user")
    public String pushToUser(String userId) {
        // 向特定用户推送消息，用户订阅了 /user/{userId}/message 才能收到
        messagingTemplate.convertAndSendToUser(userId, "/message", "您有一条私信");
        return "私信推送成功";
    }
}

<!DOCTYPE html>
<html>
<head>
    <title>STOMP over WebSocket Demo</title>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/sockjs-client/1.6.1/sockjs.min.js"></script>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/stomp.js/2.3.3/stomp.min.js"></script>
</head>
<body>
    <div>
        <input type="text" placeholder="你的名字" />
        <button onclick="sendName()">发送</button>
    </div>
    <div id="greetings"></div>
    <script>
        var stompClient = null;
        function connect() {
            var socket = new SockJS('/ws-stomp'); // 连接 SockJS 端点
            stompClient = Stomp.over(socket);
            stompClient.connect({}, function(frame) {
                console.log('连接成功：' + frame);
                // 订阅服务器的广播主题 /topic/greetings
                stompClient.subscribe('/topic/greetings', function(greeting) {
                    showGreeting(JSON.parse(greeting.body).content);
                });
                // 订阅用户的私有消息（当前用户，服务器会根据连接自动识别用户）
                stompClient.subscribe('/user/message', function(message) {
                    showGreeting("私信：" + message.body);
                });
            }, function(error) {
                console.log('连接失败：' + error);
            });
        }
        function sendName() {
            var name = document.getElementById('name').value;
            stompClient.send("/app/hello", {}, JSON.stringify({ 'name': name }));
        }
        function showGreeting(message) {
            var div = document.getElementById('greetings');
            div.innerHTML += "<p>" + message + "</p>";
        }
        // 页面加载完成后自动连接
        window.onload = connect;
    </script>
</body>
</html>

@MessageMapping("/private")
@SendToUser("/queue/private")
public String handlePrivate(String message) {
    return "这是你的私密回声：" + message;
}

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

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.web.builders.HttpSecurity;
import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
import org.springframework.security.core.userdetails.User;
import org.springframework.security.core.userdetails.UserDetailsService;
import org.springframework.security.provisioning.InMemoryUserDetailsManager;
import org.springframework.security.web.SecurityFilterChain;

@Configuration
@EnableWebSecurity
public class SecurityConfig {
    @Bean
    public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
        http
            .authorizeRequests()
            .antMatchers("/ws-stomp/**").authenticated() // WebSocket 端点需要认证
            .anyRequest().permitAll()
            .and()
            .formLogin()
            .permitAll()
            .and()
            .logout()
            .permitAll()
            .and()
            .csrf().disable(); // 简单起见禁用 CSRF，生产环境需配置
        return http.build();
    }

    @Bean
    public UserDetailsService userDetailsService() {
        InMemoryUserDetailsManager manager = new InMemoryUserDetailsManager();
        manager.createUser(User.withUsername("user1").password("{noop}password1").roles("USER").build());
        manager.createUser(User.withUsername("user2").password("{noop}password2").roles("USER").build());
        return manager;
    }
}

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

@Configuration
@EnableWebSocketMessageBroker
public class StompWebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws-stomp")
                .setAllowedOrigins("*")
                .withSockJS();
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.setApplicationDestinationPrefixes("/app");
        registry.enableSimpleBroker("/topic", "/queue");
        registry.setUserDestinationPrefix("/user");
    }

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        // 添加自定义的拦截器，用于从消息中提取用户信息
        registration.interceptors(new UserInterceptor());
    }
}

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;

public class UserInterceptor implements ChannelInterceptor {
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
        // 如果是连接帧，可以从 header 中获取 token，然后设置认证用户
        if (accessor != null && accessor.getCommand() != null) {
            switch (accessor.getCommand()) {
                case CONNECT:
                    // 假设客户端在连接时通过 header 传递了 token，例如 "Authorization: Bearer xxx"
                    String token = accessor.getFirstNativeHeader("Authorization");
                    if (token != null && token.startsWith("Bearer ")) {
                        token = token.substring(7); // 解析 token 获取用户信息，此处简化：根据 token 查数据库或 JWT 解析
                        // 假设解析出用户名为 user1
                        // 然后创建 Authentication 对象
                        // 这里简单模拟：从内存中获取用户
                        // 实际中需要从 JWT 解析出用户名，再加载 UserDetails
                        // 然后创建 UsernamePasswordAuthenticationToken
                        // 并将 authentication 设置到 SecurityContext 中
                        // 同时设置 accessor.setUser(authentication)
                        // ...
                    }
                    break;
                case SUBSCRIBE:
                    // 可以检查订阅目的地权限
                    break;
                // 其他命令...
            }
        }
        return message;
    }
}

@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
    StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
    if (accessor != null && StompCommand.SUBSCRIBE.equals(accessor.getCommand())) {
        String destination = accessor.getDestination();
        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
        if (destination != null && destination.startsWith("/topic/admin") && (auth == null || !auth.getAuthorities().contains(new SimpleGrantedAuthority("ROLE_ADMIN")))) {
            // 没有权限，拒绝订阅
            throw new AccessDeniedException("无权限订阅此主题");
        }
    }
    return message;
}

rabbitmq-plugins enable rabbitmq_stomp

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- RabbitMQ 的 STOMP 支持通常不需要额外依赖，但可能需要 spring-rabbit -->

@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
    registry.setApplicationDestinationPrefixes("/app");
    // 使用 RabbitMQ 作为外部消息代理
    registry.enableStompBrokerRelay("/topic", "/queue")
            .setRelayHost("localhost")
            .setRelayPort(61613) // STOMP 插件默认端口
            .setClientLogin("guest")
            .setClientPasscode("guest")
            .setSystemLogin("guest")
            .setSystemPasscode("guest");
    registry.setUserDestinationPrefix("/user");
}

registry.enableSimpleBroker("/topic", "/queue")
        .setHeartbeatValue(new long[]{10000, 10000}); // 客户端和服务器均每 10 秒发送一次心跳

@Override
public void configureClientInboundChannel(ChannelRegistration registration) {
    registration.taskExecutor()
            .corePoolSize(10)
            .maxPoolSize(20)
            .keepAliveSeconds(60);
}

@Override
public void configureClientOutboundChannel(ChannelRegistration registration) {
    registration.taskExecutor()
            .corePoolSize(10)
            .maxPoolSize(20)
            .keepAliveSeconds(60);
}

@Bean
public AbstractBrokerMessageHandler stompBrokerRelayMessageHandler(
        StompBrokerRelayRegistration registration,
        MessageBrokerProperties properties) {
    StompBrokerRelayMessageHandler handler = new StompBrokerRelayMessageHandler(registration);
    handler.setMessageSizeLimit(1024 * 1024); // 1MB
    return handler;
}

wscat -c ws://localhost:8080/ws/echo

import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.simp.stomp.StompFrameHandler;
import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.messaging.WebSocketStompClient;

import java.lang.reflect.Type;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class WebSocketTest {
    @Test
    void testGreeting() throws Exception {
        WebSocketStompClient stompClient = new WebSocketStompClient(new StandardWebSocketClient());
        stompClient.setMessageConverter(new MappingJackson2MessageConverter());
        String url = "ws://localhost:" + port + "/ws-stomp";
        StompSession session = stompClient.connect(url, new StompSessionHandlerAdapter() {}).get(1, TimeUnit.SECONDS);
        CompletableFuture<Greeting> future = new CompletableFuture<>();
        session.subscribe("/topic/greetings", new StompFrameHandler() {
            @Override
            public Type getPayloadType(StompHeaders headers) {
                return Greeting.class;
            }

            @Override
            public void handleFrame(StompHeaders headers, Object payload) {
                future.complete((Greeting) payload);
            }
        });
        session.send("/app/hello", new HelloMessage("TestUser"));
        Greeting greeting = future.get(5, TimeUnit.SECONDS);
        assertThat(greeting.getContent()).contains("Hello, TestUser");
    }
    // 需要注入端口
    @LocalServerPort
    private int port;
}

stompClient.connect({}, function(frame) {
    // 连接成功
}, function(error) {
    // 连接失败，延时重连
    setTimeout(connect, 5000);
});

public class Notification {
    private String type; // "BROADCAST" 或 "PRIVATE"
    private String content;
    private String targetUser; // 私信目标用户，广播时为空
    private Date timestamp;
    // 构造、getter、setter
}

@MessageMapping("/private")
public void sendPrivate(@Payload Notification notification, Principal principal) {
    // 谁发送的？可通过 principal 获取
    notification.setTimestamp(new Date());
    // 发送给目标用户
    messagingTemplate.convertAndSendToUser(notification.getTargetUser(), "/queue/notifications", notification);
}

@RestController
@RequestMapping("/api/notify")
public class NotifyController {
    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @PostMapping("/broadcast")
    public String broadcast(@RequestBody String content) {
        Notification notification = new Notification("BROADCAST", content, null, new Date());
        messagingTemplate.convertAndSend("/topic/broadcast", notification);
        return "广播成功";
    }
}

function connect() {
    var socket = new SockJS('/ws-stomp');
    stompClient = Stomp.over(socket);
    stompClient.connect({}, function(frame) {
        // 订阅广播
        stompClient.subscribe('/topic/broadcast', function(msg) {
            var notif = JSON.parse(msg.body);
            showNotification(notif.content);
        });
        // 订阅私有通知
        stompClient.subscribe('/user/queue/notifications', function(msg) {
            var notif = JSON.parse(msg.body);
            showNotification("私信：" + notif.content);
        });
    });
}

Spring Boot 集成 WebSocket 实现后台向前端推送信息

1. 引言

2. WebSocket 基础

2.1 什么是 WebSocket？

2.2 WebSocket 与 HTTP 的关系

2.3 WebSocket 的工作流程

2.4 WebSocket 的优点

3. Spring Boot 对 WebSocket 的支持

4. 准备工作：创建 Spring Boot 项目

5. 基于原生 WebSocket 的实现

5.1 配置 WebSocket 处理器

5.2 编写 WebSocket 处理类

5.3 前端 JavaScript 示例

5.4 后台主动推送消息

5.5 原生方式的优缺点

6. 基于 STOMP 的实现

6.1 STOMP 简介

6.2 配置 WebSocket 消息代理

6.3 创建 Controller 处理消息

6.4 使用 SimpMessagingTemplate 推送消息

6.5 前端使用 SockJS 和 STOMP.js

6.6 广播与点对点的深入理解

6.6.1 广播（Topic）

6.6.2 点对点（Queue）

6.6.3 使用 @SendToUser

7. 安全集成（Spring Security + WebSocket）

7.1 添加 Spring Security 依赖

7.2 配置 Spring Security

7.3 WebSocket 拦截器获取用户信息

7.4 基于角色的目的地权限控制

8. 集群环境下的 WebSocket

8.1 问题分析

8.2 解决方案概览

8.3 使用 RabbitMQ 作为外部消息代理

8.3.1 安装并启用 RabbitMQ STOMP 插件

8.3.2 修改 Spring Boot 配置

8.3.3 注意事项

8.4 使用 Redis 存储会话

9. 性能优化与最佳实践

9.1 心跳机制

9.2 线程模型

9.3 消息大小限制

9.4 连接数限制

9.5 压缩与编码

10. 测试 WebSocket

10.1 使用浏览器开发者工具

10.2 使用命令行工具

10.3 Spring Boot 集成测试

11. 常见问题与解决方案

11.1 连接失败（404）

11.2 消息丢失

11.3 断线重连

11.4 跨域问题

11.5 客户端无法接收广播消息

11.6 性能瓶颈

12. 实战案例：实时通知系统

12.1 需求描述

12.2 后端实现

12.3 前端实现

12.4 测试

13. 总结与展望

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具