Java实现消息推送案例

本文目录导读：

1. WebSocket实现方案
2. SSE（Server-Sent Events）实现方案
3. 使用示例
4. 选择建议

我将为您提供一个完整的Java消息推送实现案例,包含WebSocket和SSE两种主流方案。

WebSocket实现方案

项目依赖（Maven）

<dependencies>
    <!-- Spring Boot WebSocket -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>
    <!-- Spring Boot Web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- JSON处理 -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>2.0.32</version>
    </dependency>
</dependencies>

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;
import javax.annotation.Resource;
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Resource
    private WebSocketHandler webSocketHandler;
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(webSocketHandler, "/ws/message")
                .addInterceptors(new WebSocketInterceptor())
                .setAllowedOrigins("*");
    }
}

WebSocket处理器

import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Component
public class WebSocketHandler extends TextWebSocketHandler {
    // 存储在线会话，key: userId, value: session
    private static final Map<String, WebSocketSession> SESSIONS = new ConcurrentHashMap<>();
    /**
     * 连接建立后
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        // 从session中获取用户ID
        String userId = (String) session.getAttributes().get("userId");
        SESSIONS.put(userId, session);
        System.out.println("用户 " + userId + " 已连接");
        // 发送欢迎消息
        JSONObject welcomeMsg = new JSONObject();
        welcomeMsg.put("type", "welcome");
        welcomeMsg.put("message", "连接成功！");
        session.sendMessage(new TextMessage(welcomeMsg.toJSONString()));
    }
    /**
     * 接收消息
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        JSONObject msgObj = JSON.parseObject(payload);
        String type = msgObj.getString("type");
        switch (type) {
            case "chat":
                handleChatMessage(msgObj);
                break;
            case "ping":
                handlePing(session);
                break;
            default:
                break;
        }
    }
    /**
     * 处理聊天消息
     */
    private void handleChatMessage(JSONObject msgObj) throws Exception {
        String toUserId = msgObj.getString("toUserId");
        String content = msgObj.getString("content");
        JSONObject response = new JSONObject();
        response.put("type", "message");
        response.put("content", content);
        response.put("fromUserId", msgObj.getString("fromUserId"));
        response.put("timestamp", System.currentTimeMillis());
        // 发送给指定用户
        sendToUser(toUserId, response.toJSONString());
    }
    /**
     * 处理心跳
     */
    private void handlePing(WebSocketSession session) throws Exception {
        JSONObject pong = new JSONObject();
        pong.put("type", "pong");
        session.sendMessage(new TextMessage(pong.toJSONString()));
    }
    /**
     * 连接关闭后
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        String userId = (String) session.getAttributes().get("userId");
        SESSIONS.remove(userId);
        System.out.println("用户 " + userId + " 已断开");
    }
    /**
     * 发送消息给指定用户
     */
    public static void sendToUser(String userId, String message) throws IOException {
        WebSocketSession session = SESSIONS.get(userId);
        if (session != null && session.isOpen()) {
            session.sendMessage(new TextMessage(message));
        }
    }
    /**
     * 发送消息给所有用户（广播）
     */
    public static void sendToAll(String message) throws IOException {
        for (WebSocketSession session : SESSIONS.values()) {
            if (session.isOpen()) {
                session.sendMessage(new TextMessage(message));
            }
        }
    }
    /**
     * 获取在线用户列表
     */
    public static int getOnlineCount() {
        return SESSIONS.size();
    }
}

WebSocket拦截器（用于获取用户信息）

import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.http.server.ServletServerHttpRequest;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpSession;
import java.util.Map;
public class WebSocketInterceptor implements HandshakeInterceptor {
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
                                   WebSocketHandler handler, Map<String, Object> attributes) throws Exception {
        if (request instanceof ServletServerHttpRequest) {
            ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request;
            HttpServletRequest httpServletRequest = servletRequest.getServletRequest();
            // 从参数或Header中获取用户ID
            String userId = httpServletRequest.getParameter("userId");
            if (userId == null) {
                userId = httpServletRequest.getHeader("userId");
            }
            // 也可以从Session获取
            HttpSession session = httpServletRequest.getSession(false);
            if (session != null && session.getAttribute("userId") != null) {
                userId = (String) session.getAttribute("userId");
            }
            if (userId != null) {
                attributes.put("userId", userId);
                return true;
            }
        }
        return false;
    }
    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
                               WebSocketHandler handler, Exception exception) {
        // 握手后处理
    }
}

消息推送服务

import org.springframework.stereotype.Service;
import com.alibaba.fastjson.JSONObject;
import java.io.IOException;
@Service
public class MessagePushService {
    /**
     * 发送系统通知
     */
    public void sendSystemNotification(String userId, String title, String content) throws IOException {
        JSONObject message = new JSONObject();
        message.put("type", "system_notification");
        message.put("title", title);
        message.put("content", content);
        message.put("timestamp", System.currentTimeMillis());
        WebSocketHandler.sendToUser(userId, message.toJSONString());
    }
    /**
     * 发送业务消息
     */
    public void sendBusinessMessage(String userId, String businessType, Object data) throws IOException {
        JSONObject message = new JSONObject();
        message.put("type", "business_message");
        message.put("businessType", businessType);
        message.put("data", data);
        message.put("timestamp", System.currentTimeMillis());
        WebSocketHandler.sendToUser(userId, message.toJSONString());
    }
    /**
     * 广播消息
     */
    public void broadcast(String content) throws IOException {
        JSONObject message = new JSONObject();
        message.put("type", "broadcast");
        message.put("content", content);
        message.put("timestamp", System.currentTimeMillis());
        WebSocketHandler.sendToAll(message.toJSONString());
    }
}

控制器

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.ModelAndView;
import java.io.IOException;
@RestController
@RequestMapping("/api")
public class MessageController {
    @Autowired
    private MessagePushService messagePushService;
    /**
     * WebSocket连接页面
     */
    @GetMapping("/ws")
    public ModelAndView websocketPage() {
        ModelAndView mv = new ModelAndView("websocket");
        return mv;
    }
    /**
     * 发送系统通知
     */
    @PostMapping("/send/notification")
    public String sendNotification(@RequestParam String userId,
                                  @RequestParam String title,
                                  @RequestParam String content) throws IOException {
        messagePushService.sendSystemNotification(userId, title, content);
        return "发送成功";
    }
    /**
     * 发送业务消息
     */
    @PostMapping("/send/business")
    public String sendBusinessMessage(@RequestParam String userId,
                                      @RequestParam String businessType,
                                      @RequestBody Object data) throws IOException {
        messagePushService.sendBusinessMessage(userId, businessType, data);
        return "发送成功";
    }
    /**
     * 广播消息
     */
    @PostMapping("/broadcast")
    public String broadcast(@RequestParam String content) throws IOException {
        messagePushService.broadcast(content);
        return "广播成功";
    }
    /**
     * 获取在线人数
     */
    @GetMapping("/online-count")
    public int getOnlineCount() {
        return WebSocketHandler.getOnlineCount();
    }
}

前端页面（websocket.html）

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">WebSocket 消息推送</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; }
        .message-box { width: 100%; height: 300px; border: 1px solid #ccc; 
                      overflow-y: auto; padding: 10px; margin-bottom: 20px; }
        .message { margin-bottom: 10px; padding: 5px; }
        .received { background-color: #e3f2fd; }
        .sent { background-color: #f1f8e9; }
        .input-area { display: flex; gap: 10px; }
        input[type="text"] { flex: 1; padding: 8px; }
        button { padding: 8px 20px; background-color: #007bff; color: white; 
                border: none; border-radius: 4px; cursor: pointer; }
    </style>
</head>
<body>
    <h1>WebSocket 消息推送测试</h1>
    <div>
        <label>用户ID：<input type="text" id="userId" value="user001" readonly></label>
        <label>在线人数：<span id="onlineCount">0</span></label>
    </div>
    <div class="message-box" id="messageBox"></div>
    <div class="input-area">
        <input type="text" id="messageInput" placeholder="输入消息内容...">
        <button onclick="sendMessage()">发送</button>
        <button onclick="connect()">重新连接</button>
        <button onclick="disconnect()">断开连接</button>
    </div>
    <script>
        let ws = null;
        const userId = document.getElementById('userId').value;
        function connect() {
            if (ws && ws.readyState !== WebSocket.CLOSED) return;
            // WebSocket 连接
            ws = new WebSocket(`ws://localhost:8080/ws/message?userId=${userId}`);
            ws.onopen = function() {
                addMessage('系统', '连接成功', 'sent');
            };
            ws.onmessage = function(event) {
                const data = JSON.parse(event.data);
                addMessage('服务器', JSON.stringify(data), 'received');
                // 处理不同类型消息
                switch(data.type) {
                    case 'system_notification':
                        alert('系统通知：' + data.content);
                        break;
                    case 'broadcast':
                        console.log('广播消息：' + data.content);
                        break;
                    default:
                        break;
                }
            };
            ws.onerror = function(error) {
                addMessage('系统', '连接错误', 'received');
            };
            ws.onclose = function() {
                addMessage('系统', '连接关闭', 'received');
            };
        }
        function sendMessage() {
            const input = document.getElementById('messageInput');
            const content = input.value.trim();
            if (!content || !ws || ws.readyState !== WebSocket.OPEN) {
                alert('请先连接或输入内容');
                return;
            }
            const message = {
                type: 'chat',
                fromUserId: userId,
                toUserId: 'server',
                content: content
            };
            ws.send(JSON.stringify(message));
            addMessage('我', content, 'sent');
            input.value = '';
        }
        function disconnect() {
            if (ws) {
                ws.close();
            }
        }
        function addMessage(from, content, type) {
            const messageBox = document.getElementById('messageBox');
            const div = document.createElement('div');
            div.className = `message ${type}`;
            div.innerHTML = `<strong>${from}:</strong> ${content}`;
            messageBox.appendChild(div);
            messageBox.scrollTop = messageBox.scrollHeight;
        }
        function updateOnlineCount(count) {
            document.getElementById('onlineCount').textContent = count;
        }
        // 页面加载时自动连接
        window.onload = function() {
            connect();
            // 定时获取在线人数
            setInterval(function() {
                fetch('/api/online-count')
                    .then(response => response.json())
                    .then(count => updateOnlineCount(count))
                    .catch(console.error);
            }, 5000);
        };
    </script>
</body>
</html>

SSE（Server-Sent Events）实现方案

SSE控制器

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@RestController
@RequestMapping("/sse")
public class SseController {
    // 存储SSE连接，key: userId
    private static final Map<String, SseEmitter> EMITTERS = new ConcurrentHashMap<>();
    /**
     * 建立SSE连接
     */
    @GetMapping(value = "/connect/{userId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter connect(@PathVariable String userId) {
        SseEmitter emitter = new SseEmitter(0L); // 0L表示不超时
        // 存储连接
        EMITTERS.put(userId, emitter);
        // 连接完成时回调
        emitter.onCompletion(() -> {
            EMITTERS.remove(userId);
            System.out.println("SSE连接完成：" + userId);
        });
        // 连接超时回调
        emitter.onTimeout(() -> {
            EMITTERS.remove(userId);
            System.out.println("SSE连接超时：" + userId);
        });
        // 连接错误回调
        emitter.onError((error) -> {
            EMITTERS.remove(userId);
            System.out.println("SSE连接错误：" + userId);
        });
        try {
            // 发送初始消息
            emitter.send(SseEmitter.event().name("connect").data("连接成功"));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return emitter;
    }
    /**
     * 发送消息
     */
    @PostMapping("/send/{userId}")
    public String sendEvent(@PathVariable String userId, @RequestBody String message) {
        SseEmitter emitter = EMITTERS.get(userId);
        if (emitter != null) {
            try {
                emitter.send(SseEmitter.event().name("message").data(message));
                return "消息发送成功";
            } catch (IOException e) {
                e.printStackTrace();
                return "发送失败";
            }
        }
        return "用户不在线";
    }
    /**
     * 广播消息
     */
    @PostMapping("/broadcast")
    public String broadcast(@RequestBody String message) {
        int successCount = 0;
        for (Map.Entry<String, SseEmitter> entry : EMITTERS.entrySet()) {
            try {
                entry.getValue().send(SseEmitter.event().name("broadcast").data(message));
                successCount++;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return "广播给 " + successCount + " 个用户";
    }
    /**
     * 关闭连接
     */
    @GetMapping("/disconnect/{userId}")
    public String disconnect(@PathVariable String userId) {
        SseEmitter emitter = EMITTERS.remove(userId);
        if (emitter != null) {
            emitter.complete();
            return "连接已关闭";
        }
        return "连接不存在";
    }
}

SSE前端页面（sse.html）

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">SSE 消息推送</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; }
        .message-box { width: 100%; height: 300px; border: 1px solid #ccc; 
                      overflow-y: auto; padding: 10px; margin-top: 20px; }
        button { padding: 8px 20px; background-color: #28a745; color: white; 
                border: none; border-radius: 4px; cursor: pointer; }
        input { padding: 8px; margin-right: 10px; }
    </style>
</head>
<body>
    <h1>SSE 消息推送测试</h1>
    <div>
        <input type="text" id="userId" value="user001" placeholder="用户ID">
        <button onclick="connect()">连接</button>
        <button onclick="disconnect()">断开</button>
    </div>
    <div class="message-box" id="messageBox"></div>
    <script>
        let eventSource = null;
        function connect() {
            const userId = document.getElementById('userId').value;
            if (eventSource) {
                eventSource.close();
            }
            // 建立SSE连接
            eventSource = new EventSource(`/sse/connect/${userId}`);
            // 连接打开
            eventSource.onopen = function() {
                addMessage('系统', 'SSE连接已建立', 'received');
            };
            // 监听custom事件
            eventSource.addEventListener('message', function(event) {
                addMessage('服务器', event.data, 'received');
            });
            eventSource.addEventListener('broadcast', function(event) {
                addMessage('广播', event.data, 'received');
            });
            eventSource.addEventListener('connect', function(event) {
                addMessage('服务器', event.data, 'received');
            });
            // 错误处理
            eventSource.onerror = function(event) {
                addMessage('系统', '连接发生错误，重试中...', 'received');
            };
        }
        function disconnect() {
            if (eventSource) {
                eventSource.close();
                addMessage('系统', '连接已断开', 'received');
            }
        }
        function addMessage(from, content, type) {
            const messageBox = document.getElementById('messageBox');
            const div = document.createElement('div');
            div.innerHTML = `<strong>${from}:</strong> ${content}`;
            messageBox.appendChild(div);
            messageBox.scrollTop = messageBox.scrollHeight;
        }
        // 页面加载时自动连接
        window.onload = function() {
            connect();
        };
    </script>
</body>
</html>

使用示例

发送系统通知（WebSocket）

curl -X POST \
  http://localhost:8080/api/send/notification \
  -d 'userId=user001&title=系统通知&content=您的订单已发货' \
  -H 'Content-Type: application/x-www-form-urlencoded'

广播消息（SSE）

curl -X POST \
  http://localhost:8080/sse/broadcast \
  -H 'Content-Type: text/plain' \
  -d '欢迎参加双十一活动！'

选择建议

1. WebSocket适用场景：

   • 需要双向通信
   • 高频消息推送
   • 实时聊天、游戏等

2. SSE适用场景：

   • 服务端到客户端的单向推送
   • 实时通知、股票行情
   • 自动重连机制

3. 长轮询（AJAX Polling）：

   • 简单场景
   • 兼容性要求高
   • 低频更新

这个案例包含了完整的实现代码和使用示例，您可以根据实际需求选择合适的方案，需要添加更多功能（如消息持久化、群组推送等）,可以在此基础上扩展。