🌐 Java WebSocket 全面解析：构建企业级实时通信系统

协议基础与Java实现标准

Java中的WebSocket支持

Java提供了两种主要的WebSocket实现方式：

JSR356标准API：Java EE 7及以后版本的标准WebSocket API
Spring Framework：基于Spring的简化WebSocket实现
底层API：使用Java NIO和自定义协议实现

环境要求

<!-- Maven 依赖配置 -->
<dependencies>
    <!-- Jakarta WebSocket API -->
    <dependency>
        <groupId>jakarta.websocket</groupId>
        <artifactId>jakarta.websocket-api</artifactId>
        <version>2.1.0</version>
    </dependency>
    
    <!-- Tomcat WebSocket 实现 -->
    <dependency>
        <groupId>org.apache.tomcat.embed</groupId>
        <artifactId>tomcat-embed-websocket</artifactId>
        <version>10.1.5</version>
    </dependency>
    
    <!-- 或者使用 Jetty -->
    <dependency>
        <groupId>org.eclipse.jetty.websocket</groupId>
        <artifactId>websocket-jakarta-server</artifactId>
        <version>11.0.15</version>
    </dependency>
</dependencies>

JSR356 WebSocket API 详解

服务端端点基础实现

import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 基础WebSocket服务器端点
 * 使用@ServerEndpoint注解声明WebSocket端点
 */
@ServerEndpoint(
    value = "/chat/{roomId}",  // 路径参数支持
    encoders = MessageEncoder.class,  // 消息编码器
    decoders = MessageDecoder.class,  // 消息解码器
    configurator = WebSocketConfigurator.class  // 配置器
)
public class ChatEndpoint {
    
    // 线程安全的连接集合
    private static final Set<Session> sessions = 
        Collections.newSetFromMap(new ConcurrentHashMap<>());
    
    private String roomId;
    private Session session;
    
    /**
     * 连接建立时调用
     * @param session WebSocket会话
     * @param config 端点配置
     */
    @OnOpen
    public void onOpen(Session session, EndpointConfig config) {
        this.session = session;
        this.roomId = session.getPathParameters().get("roomId");
        
        sessions.add(session);
        System.out.println("新连接建立: " + session.getId() + ", 房间: " + roomId);
        
        // 发送欢迎消息
        sendMessage(session, new Message("system", "欢迎加入聊天室 " + roomId));
    }
    
    /**
     * 接收文本消息
     * @param message 接收到的消息
     * @param session 发送消息的会话
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        try {
            Message msg = MessageDecoder.decode(message);
            broadcastMessage(msg);
        } catch (IOException e) {
            System.err.println("消息处理失败: " + e.getMessage());
        }
    }
    
    /**
     * 接收二进制消息
     * @param bytes 二进制数据
     * @param session 会话对象
     */
    @OnMessage
    public void onBinary(byte[] bytes, Session session) {
        // 处理二进制消息
        System.out.println("收到二进制数据, 长度: " + bytes.length);
    }
    
    /**
     * 连接关闭时调用
     * @param session 关闭的会话
     * @param closeReason 关闭原因
     */
    @OnClose
    public void onClose(Session session, CloseReason closeReason) {
        sessions.remove(session);
        System.out.println("连接关闭: " + session.getId() + 
                          ", 原因: " + closeReason.getReasonPhrase());
    }
    
    /**
     * 发生错误时调用
     * @param session 发生错误的会话
     * @param throwable 异常信息
     */
    @OnError
    public void onError(Session session, Throwable throwable) {
        System.err.println("WebSocket错误: " + throwable.getMessage());
        try {
            session.close(new CloseReason(
                CloseReason.CloseCodes.UNEXPECTED_CONDITION, 
                "服务器错误"
            ));
        } catch (IOException e) {
            System.err.println("关闭连接时发生错误: " + e.getMessage());
        }
    }
    
    /**
     * 广播消息给所有连接
     * @param message 要广播的消息
     */
    private void broadcastMessage(Message message) {
        sessions.forEach(session -> {
            if (session.isOpen()) {
                sendMessage(session, message);
            }
        });
    }
    
    /**
     * 发送消息到指定会话
     * @param session 目标会话
     * @param message 消息对象
     */
    private void sendMessage(Session session, Message message) {
        try {
            session.getBasicRemote().sendText(MessageEncoder.encode(message));
        } catch (IOException e) {
            System.err.println("消息发送失败: " + e.getMessage());
        }
    }
}

/**
 * 消息实体类
 */
public class Message {
    private String from;
    private String content;
    private long timestamp;
    private MessageType type;
    
    public enum MessageType {
        TEXT, IMAGE, FILE, SYSTEM
    }
    
    // 构造函数、getter、setter省略
}

/**
 * 消息编码器
 */
public class MessageEncoder implements Encoder.Text<Message> {
    @Override
    public String encode(Message message) throws EncodeException {
        return JsonUtil.toJson(message);  // 使用JSON序列化
    }
    
    @Override
    public void init(EndpointConfig config) {}
    
    @Override
    public void destroy() {}
}

/**
 * 消息解码器
 */
public class MessageDecoder implements Decoder.Text<Message> {
    @Override
    public Message decode(String text) throws DecodeException {
        return JsonUtil.fromJson(text, Message.class);
    }
    
    @Override
    public boolean willDecode(String text) {
        try {
            JsonUtil.fromJson(text, Message.class);
            return true;
        } catch (Exception e) {
            return false;
        }
    }
    
    @Override
    public void init(EndpointConfig config) {}
    
    @Override
    public void destroy() {}
}

客户端端点实现

import jakarta.websocket.*;
import java.net.URI;

/**
 * WebSocket客户端端点
 */
@ClientEndpoint(
    encoders = MessageEncoder.class,
    decoders = MessageDecoder.class,
    configurator = ClientWebSocketConfigurator.class
)
public class ChatClientEndpoint {
    
    private Session session;
    private MessageHandler messageHandler;
    
    public ChatClientEndpoint(URI endpointURI) {
        try {
            WebSocketContainer container = ContainerProvider.getWebSocketContainer();
            container.connectToServer(this, endpointURI);
        } catch (Exception e) {
            throw new RuntimeException("连接WebSocket服务器失败", e);
        }
    }
    
    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
        System.out.println("连接已建立: " + session.getId());
    }
    
    @OnMessage
    public void onMessage(String message) {
        if (messageHandler != null) {
            messageHandler.handleMessage(message);
        }
    }
    
    @OnClose
    public void onClose(Session session, CloseReason closeReason) {
        System.out.println("连接关闭: " + closeReason.getReasonPhrase());
    }
    
    public void addMessageHandler(MessageHandler handler) {
        this.messageHandler = handler;
    }
    
    public void sendMessage(Message message) {
        try {
            session.getBasicRemote().sendObject(message);
        } catch (Exception e) {
            System.err.println("发送消息失败: " + e.getMessage());
        }
    }
    
    public interface MessageHandler {
        void handleMessage(String message);
    }
}

Spring Boot WebSocket 集成

Spring WebSocket 配置

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;

/**
 * Spring WebSocket配置类
 */
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    
    private final ChatWebSocketHandler chatHandler;
    private final AuthHandshakeInterceptor authInterceptor;
    
    public WebSocketConfig(ChatWebSocketHandler chatHandler, 
                          AuthHandshakeInterceptor authInterceptor) {
        this.chatHandler = chatHandler;
        this.authInterceptor = authInterceptor;
    }
    
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(chatHandler, "/ws/chat")
                .addInterceptors(authInterceptor)
                .setAllowedOrigins("*");  // 生产环境应限制来源
        
        // 支持SockJS降级方案
        registry.addHandler(chatHandler, "/ws/chat/sockjs")
                .addInterceptors(authInterceptor)
                .withSockJS();
    }
}

/**
 * WebSocket处理器
 */
@Component
public class ChatWebSocketHandler extends TextWebSocketHandler {
    
    private final SimpMessagingTemplate messagingTemplate;
    private final SessionManager sessionManager;
    
    public ChatWebSocketHandler(SimpMessagingTemplate messagingTemplate,
                               SessionManager sessionManager) {
        this.messagingTemplate = messagingTemplate;
        this.sessionManager = sessionManager;
    }
    
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        String username = session.getAttributes().get("username").toString();
        sessionManager.addSession(username, session);
        
        // 通知用户上线
        messagingTemplate.convertAndSend("/topic/online", 
            new OnlineEvent(username, true));
    }
    
    @Override
    protected void handleTextMessage(WebSocketSession session, 
                                   TextMessage message) throws Exception {
        ChatMessage chatMessage = objectMapper.readValue(
            message.getPayload(), ChatMessage.class);
        
        // 消息验证和处理
        if (validateMessage(chatMessage)) {
            messagingTemplate.convertAndSendToUser(
                chatMessage.getTo(),
                "/queue/messages",
                chatMessage
            );
        }
    }
    
    @Override
    public void afterConnectionClosed(WebSocketSession session, 
                                    CloseStatus status) {
        String username = session.getAttributes().get("username").toString();
        sessionManager.removeSession(username);
        
        // 通知用户下线
        messagingTemplate.convertAndSend("/topic/online",
            new OnlineEvent(username, false));
    }
    
    private boolean validateMessage(ChatMessage message) {
        // 实现消息验证逻辑
        return message != null && 
               message.getContent() != null &&
               !message.getContent().trim().isEmpty();
    }
}

/**
 * 握手拦截器 - 用于身份验证
 */
@Component
public class AuthHandshakeInterceptor implements HandshakeInterceptor {
    
    private final JwtTokenProvider tokenProvider;
    
    @Override
    public boolean beforeHandshake(ServerHttpRequest request,
                                  ServerHttpResponse response,
                                  WebSocketHandler wsHandler,
                                  Map<String, Object> attributes) throws Exception {
        
        // 从请求参数中获取token
        String token = getTokenFromRequest(request);
        
        if (token != null && tokenProvider.validateToken(token)) {
            String username = tokenProvider.getUsernameFromToken(token);
            attributes.put("username", username);
            return true;
        }
        
        response.setStatusCode(HttpStatus.UNAUTHORIZED);
        return false;
    }
    
    @Override
    public void afterHandshake(ServerHttpRequest request,
                              ServerHttpResponse response,
                              WebSocketHandler wsHandler,
                              Exception exception) {
        // 握手完成后执行
    }
    
    private String getTokenFromRequest(ServerHttpRequest request) {
        // 从请求中提取token的逻辑
        return null;
    }
}

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;

/**
 * STOMP WebSocket配置
 */
@Configuration
@EnableWebSocketMessageBroker
public class StompWebSocketConfig implements WebSocketMessageBrokerConfigurer {
    
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        // 启用简单的内存消息代理
        config.enableSimpleBroker("/topic", "/queue");
        
        // 配置应用程序目的地前缀
        config.setApplicationDestinationPrefixes("/app");
        
        // 用户目的地前缀
        config.setUserDestinationPrefix("/user");
    }
    
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
                .setAllowedOriginPatterns("*")
                .withSockJS();
        
        registry.addEndpoint("/ws")
                .setAllowedOriginPatterns("*");
    }
    
    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
        registration.setMessageSizeLimit(128 * 1024); // 128KB
        registration.setSendTimeLimit(10000); // 10秒
        registration.setSendBufferSizeLimit(512 * 1024); // 512KB
    }
}

高级特性与优化策略

连接管理与心跳机制

/**
 * WebSocket会话管理器
 */
@Component
public class SessionManager {
    
    private final ConcurrentMap<String, WebSocketSession> sessions = 
        new ConcurrentHashMap<>();
    
    private final ScheduledExecutorService heartbeatScheduler =
        Executors.newSingleThreadScheduledExecutor();
    
    @PostConstruct
    public void init() {
        // 每30秒执行一次心跳检查
        heartbeatScheduler.scheduleAtFixedRate(
            this::checkHeartbeats, 30, 30, TimeUnit.SECONDS);
    }
    
    public void addSession(String userId, WebSocketSession session) {
        sessions.put(userId, session);
        session.getAttributes().put("lastHeartbeat", System.currentTimeMillis());
    }
    
    public void updateHeartbeat(String userId) {
        WebSocketSession session = sessions.get(userId);
        if (session != null) {
            session.getAttributes().put("lastHeartbeat", System.currentTimeMillis());
        }
    }
    
    private void checkHeartbeats() {
        long currentTime = System.currentTimeMillis();
        long timeout = 120 * 1000; // 2分钟超时
        
        sessions.entrySet().removeIf(entry -> {
            WebSocketSession session = entry.getValue();
            Long lastHeartbeat = (Long) session.getAttributes().get("lastHeartbeat");
            
            if (lastHeartbeat == null || currentTime - lastHeartbeat > timeout) {
                try {
                    session.close(CloseStatus.SESSION_NOT_RELIABLE);
                } catch (IOException e) {
                    // 记录日志
                }
                return true;
            }
            return false;
        });
    }
    
    public void broadcast(String destination, Object payload) {
        sessions.values().forEach(session -> {
            if (session.isOpen()) {
                try {
                    session.sendMessage(new TextMessage(
                        objectMapper.writeValueAsString(payload)
                    ));
                } catch (IOException e) {
                    // 处理发送失败
                }
            }
        });
    }
}

消息压缩与性能优化

/**
 * 消息压缩工具类
 */
public class MessageCompressionUtil {
    
    private static final int COMPRESSION_THRESHOLD = 1024; // 1KB
    
    public static byte[] compressIfNeeded(byte[] data) throws IOException {
        if (data.length < COMPRESSION_THRESHOLD) {
            return data;
        }
        
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        try (DeflaterOutputStream compressor = 
             new DeflaterOutputStream(outputStream)) {
            compressor.write(data);
        }
        return outputStream.toByteArray();
    }
    
    public static byte[] decompress(byte[] compressedData) throws IOException {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        try (InflaterInputStream decompressor = 
             new InflaterInputStream(new ByteArrayInputStream(compressedData))) {
            byte[] buffer = new byte[1024];
            int bytesRead;
            while ((bytesRead = decompressor.read(buffer)) != -1) {
                outputStream.write(buffer, 0, bytesRead);
            }
        }
        return outputStream.toByteArray();
    }
}

/**
 * 自定义WebSocket消息压缩器
 */
public class CompressingEncoder implements Encoder.Binary<Message> {
    
    @Override
    public ByteBuffer encode(Message message) throws EncodeException {
        try {
            byte[] jsonData = JsonUtil.toJson(message).getBytes(StandardCharsets.UTF_8);
            byte[] compressed = MessageCompressionUtil.compressIfNeeded(jsonData);
            return ByteBuffer.wrap(compressed);
        } catch (IOException e) {
            throw new EncodeException(message, "压缩失败", e);
        }
    }
    
    @Override
    public void init(EndpointConfig config) {}
    
    @Override
    public void destroy() {}
}

安全实践与生产部署

安全配置

/**
 * WebSocket安全配置
 */
@Configuration
public class WebSocketSecurityConfig {
    
    @Bean
    public DefaultHandshakeHandler handshakeHandler() {
        WebSocketPolicy policy = new WebSocketPolicy(WebSocketBehavior.SERVER);
        policy.setInputBufferSize(8192);
        policy.setIdleTimeout(600000);
        
        return new DefaultHandshakeHandler(
            new JettyWebSocketServerFactory(policy));
    }
    
    @Bean
    public HandshakeInterceptor authorizationInterceptor() {
        return new HandshakeInterceptor() {
            @Override
            public boolean beforeHandshake(ServerHttpRequest request,
                                         ServerHttpResponse response,
                                         WebSocketHandler wsHandler,
                                         Map<String, Object> attributes) {
                // JWT token验证
                String token = extractToken(request);
                return validateJwtToken(token);
            }
            
            // 其他方法实现...
        };
    }
}

/**
 * 消息内容安全检查
 */
@Component
public class MessageContentFilter {
    
    private final AntiSamy antiSamy;
    private final ProfanityFilter profanityFilter;
    
    public MessageContentFilter() throws PolicyException {
        this.antiSamy = new AntiSamy();
        this.profanityFilter = new ProfanityFilter();
    }
    
    public String sanitize(String input) throws ScanException, IOException {
        // XSS防护
        CleanResults cleanResults = antiSamy.scan(input);
        String sanitized = cleanResults.getCleanHTML();
        
        // 敏感词过滤
        return profanityFilter.filter(sanitized);
    }
}

集群部署方案

/**
 * 基于Redis的分布式WebSocket消息广播
 */
@Component
public class RedisWebSocketMessageBroker {
    
    private final RedisTemplate<String, Object> redisTemplate;
    private final SimpMessagingTemplate messagingTemplate;
    private final MessageConverter messageConverter;
    
    public RedisWebSocketMessageBroker(RedisTemplate<String, Object> redisTemplate,
                                      SimpMessagingTemplate messagingTemplate) {
        this.redisTemplate = redisTemplate;
        this.messagingTemplate = messagingTemplate;
        this.messageConverter = new Jackson2JsonMessageConverter();
    }
    
    @PostConstruct
    public void init() {
        // 订阅Redis频道
        redisTemplate.getConnectionFactory().getConnection().subscribe(
            (message, pattern) -> {
                try {
                    BroadcastMessage broadcast = (BroadcastMessage) 
                        messageConverter.fromMessage(
                            new org.springframework.data.redis.connection.DefaultMessage(
                                message.getBody(), message.getChannel()),
                            BroadcastMessage.class);
                    
                    // 本地广播消息
                    messagingTemplate.convertAndSend(
                        broadcast.getDestination(), 
                        broadcast.getPayload());
                } catch (Exception e) {
                    // 处理异常
                }
            },
            "websocket:broadcast".getBytes()
        );
    }
    
    public void broadcast(String destination, Object payload) {
        BroadcastMessage message = new BroadcastMessage(destination, payload);
        redisTemplate.convertAndSend("websocket:broadcast", message);
    }
    
    @Data
    @AllArgsConstructor
    private static class BroadcastMessage {
        private String destination;
        private Object payload;
    }
}

性能监控与故障排除

监控指标收集

/**
 * WebSocket性能监控
 */
@Component
public class WebSocketMetrics {
    
    private final MeterRegistry meterRegistry;
    private final ConcurrentMap<String, AtomicInteger> connectionCounts = 
        new ConcurrentHashMap<>();
    
    public WebSocketMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        initMetrics();
    }
    
    private void initMetrics() {
        // 连接数指标
        Gauge.builder("websocket.connections", connectionCounts::size)
             .description("当前WebSocket连接数")
             .register(meterRegistry);
        
        // 消息率指标
        Meter.builder("websocket.messages.received")
             .description("接收的消息数量")
             .register(meterRegistry);
    }
    
    public void recordConnectionEstablished(String endpoint) {
        connectionCounts.putIfAbsent(endpoint, new AtomicInteger(0));
        connectionCounts.get(endpoint).incrementAndGet();
        
        meterRegistry.counter("websocket.connections.established",
            "endpoint", endpoint).increment();
    }
    
    public void recordMessageReceived(int size) {
        meterRegistry.summary("websocket.messages.size").record(size);
        meterRegistry.counter("websocket.messages.received").increment();
    }
}

/**
 * 健康检查端点
 */
@Component
public class WebSocketHealthIndicator implements HealthIndicator {
    
    private final SessionManager sessionManager;
    
    @Override
    public Health health() {
        int activeConnections = sessionManager.getActiveConnectionCount();
        long memoryUsage = getMemoryUsage();
        
        Health.Builder builder = Health.up()
            .withDetail("activeConnections", activeConnections)
            .withDetail("memoryUsage", memoryUsage);
        
        if (memoryUsage > 80) {
            builder = Health.down()
                .withDetail("reason", "内存使用率过高: " + memoryUsage + "%");
        }
        
        return builder.build();
    }
    
    private long getMemoryUsage() {
        Runtime runtime = Runtime.getRuntime();
        long totalMemory = runtime.totalMemory();
        long freeMemory = runtime.freeMemory();
        return (totalMemory - freeMemory) * 100 / totalMemory;
    }
}

日志与故障排除

/**
 * WebSocket详细日志记录
 */
@Aspect
@Component
@Slf4j
public class WebSocketLoggingAspect {
    
    @Pointcut("within(@jakarta.websocket.server.ServerEndpoint *)")
    public void websocketEndpoint() {}
    
    @Around("websocketEndpoint() && execution(* *.*(..))")
    public Object logWebSocketOperations(ProceedingJoinPoint joinPoint) throws Throwable {
        String methodName = joinPoint.getSignature().getName();
        Object[] args = joinPoint.getArgs();
        
        if (args.length > 0 && args[0] instanceof Session) {
            Session session = (Session) args[0];
            log.debug("WebSocket操作: {} - Session: {}", methodName, session.getId());
        }
        
        long startTime = System.currentTimeMillis();
        try {
            Object result = joinPoint.proceed();
            long duration = System.currentTimeMillis() - startTime;
            
            log.debug("WebSocket操作完成: {} - 耗时: {}ms", methodName, duration);
            return result;
        } catch (Exception e) {
            log.error("WebSocket操作失败: {} - 错误: {}", methodName, e.getMessage(), e);
            throw e;
        }
    }
}

/**
 * 异常处理策略
 */
@ControllerAdvice
public class WebSocketExceptionHandler {
    
    @MessageExceptionHandler
    @SendToUser("/queue/errors")
    public ErrorMessage handleException(Exception ex) {
        log.error("WebSocket消息处理异常", ex);
        
        return new ErrorMessage(
            "处理失败",
            ex.getMessage(),
            System.currentTimeMillis()
        );
    }
    
    @Data
    @AllArgsConstructor
    public static class ErrorMessage {
        private String type;
        private String message;
        private long timestamp;
    }
}

总结与最佳实践

🎯 核心总结

Java WebSocket 实现提供了强大而灵活的实时通信能力，通过 JSR356 标准和 Spring Framework 的支持，可以构建出企业级的实时应用系统。

关键技术要点

协议层：深入理解 WebSocket 握手过程和帧格式
API选择：根据需求选择 JSR356 或 Spring WebSocket
性能优化：连接池管理、消息压缩、心跳机制
安全防护：身份验证、输入过滤、XSS防护
集群部署：分布式消息广播、会话复制

性能指标参考

📊 性能基准参考

场景	连接数	消息频率	内存占用	建议配置
小型应用	1,000	10 msg/s	512MB	单节点
中型应用	10,000	100 msg/s	2GB	2节点集群
大型应用	100,000	1,000 msg/s	8GB	5+节点集群
超大规模	1,000,000+	10,000+ msg/s	32GB+	自动扩展集群

🛡️ 安全最佳实践

始终验证：对所有输入数据进行严格验证
使用WSS：生产环境强制使用加密连接
限制资源：设置合适的消息大小和连接超时限制
监控审计：记录所有关键操作和安全事件
定期更新：保持依赖库的安全更新

🚀 部署建议

使用反向代理：Nginx 或 HAProxy 进行负载均衡
会话亲和性：配置合适的会话保持策略
健康检查：实现完善的健康检查机制
日志集中：使用 ELK 或类似方案集中管理日志
监控告警：设置性能监控和自动告警

通过遵循这些最佳实践，基于 Java 的 WebSocket 实现能够为企业应用提供稳定、安全、高效的实时通信能力，满足各种复杂的业务场景需求。

注意：本文示例基于 Java 17 和 Spring Boot 3.x，实际使用时请根据您的具体环境进行调整。生产环境部署前请进行充分的性能测试和安全审计。