java中消息推送功能的关键实现代码

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   业务系统API   │───>│   消息路由模块   │───>│   连接管理模块   │───> 客户端
└─────────────────┘    └─────────────────┘    └─────────────────┘
        │                       │                       │
        ▼                       ▼                       ▼
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│  离线消息存储    │<───│  集群通信模块    │<───│  会话管理模块    │
└─────────────────┘    └─────────────────┘    └─────────────────┘

<!-- WebSocket核心依赖 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<!-- Redis（集群通信与会话存储） -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Autowired
    private MessageHandler messageHandler; // 自定义消息处理器

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        // 配置WebSocket端点：客户端通过ws://ip:port/ws?token=xxx连接
        registry.addHandler(messageHandler, "/ws")
                .setAllowedOrigins("*") // 允许跨域（生产环境需限制域名）
                .addInterceptors(new HandshakeInterceptor() {
                    // 握手前验证用户（如Token解析用户ID）
                    @Override
                    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, 
                                                  WebSocketHandler wsHandler, Map<String, Object> attributes) {
                        String token = ((ServletServerHttpRequest) request).getServletRequest().getParameter("token");
                        String userId = parseToken(token); // 解析Token获取用户ID
                        if (userId == null) {
                            response.setStatusCode(HttpStatus.UNAUTHORIZED);
                            return false; // 验证失败，拒绝连接
                        }
                        attributes.put("userId", userId); // 存储用户ID到属性中
                        return true;
                    }

                    @Override
                    public void afterHandshake(...) {}
                });
    }
}

@Component
public class MessageHandler extends TextWebSocketHandler {
    // 本地会话映射：用户ID -> WebSocket会话（单机用）
    private final Map<String, WebSocketSession> localSessions = new ConcurrentHashMap<>();
    @Autowired
    private StringRedisTemplate redisTemplate; // Redis操作模板
    @Autowired
    private OfflineMessageService offlineMessageService; // 离线消息服务

    // 连接建立时：绑定用户ID与Session
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        String userId = (String) session.getAttributes().get("userId");
        localSessions.put(userId, session);
        
        // 1. 拉取离线消息并推送
        List<Message> offlineMessages = offlineMessageService.getByUserId(userId);
        for (Message msg : offlineMessages) {
            sendMessage(session, msg.getContent());
        }
        offlineMessageService.deleteByUserId(userId); // 清除已推送的离线消息
        
        // 2. 集群：将用户ID与当前节点IP存入Redis（便于其他节点定位）
        redisTemplate.opsForValue().set("user:session:" + userId, getLocalIp());
    }

    // 连接关闭时：移除会话映射
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        String userId = (String) session.getAttributes().get("userId");
        localSessions.remove(userId);
        redisTemplate.delete("user:session:" + userId); // 集群：删除Redis映射
    }

    // 发送消息到指定用户（核心方法）
    public void pushToUser(String userId, String content) {
        // 1. 先检查本地是否有该用户的连接
        WebSocketSession session = localSessions.get(userId);
        if (session != null && session.isOpen()) {
            sendMessage(session, content);
            return;
        }

        // 2. 集群：检查用户是否连接在其他节点
        String targetNodeIp = redisTemplate.opsForValue().get("user:session:" + userId);
        if (targetNodeIp != null) {
            // 发布消息到Redis频道，目标节点订阅后发送
            redisTemplate.convertAndSend("push:channel", 
                JSON.toJSONString(new PushMessage(userId, content)));
            return;
        }

        // 3. 用户不在线：存入离线消息
        offlineMessageService.save(new Message(userId, content, LocalDateTime.now()));
    }

    // 实际发送消息（封装异常处理）
    private void sendMessage(WebSocketSession session, String content) {
        try {
            session.sendMessage(new TextMessage(content));
        } catch (IOException e) {
            log.error("消息发送失败", e);
        }
    }
}

@Component
public class RedisMessageListener {
    @Autowired
    private MessageHandler messageHandler;

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 订阅推送频道，接收其他节点的消息
        container.addMessageListener((message, pattern) -> {
            String json = new String(message.getBody());
            PushMessage pushMsg = JSON.parseObject(json, PushMessage.class);
            // 调用本地消息处理器发送（此时用户应在当前节点）
            messageHandler.pushToUser(pushMsg.getUserId(), pushMsg.getContent());
        }, new ChannelTopic("push:channel"));
        return container;
    }
}

@RestController
@RequestMapping("/api/message")
public class MessageController {
    @Autowired
    private MessageHandler messageHandler;

    // 业务系统调用此接口触发推送
    @PostMapping("/push")
    public Result push(@RequestBody PushRequest request) {
        messageHandler.pushToUser(request.getUserId(), request.getContent());
        return Result.success();
    }
}