引言
在Web应用开发中，实时消息推送已经成为提升用户体验的重要功能。无论是聊天应用、实时通知、股票价格更新，还是系统状态监控，都需要服务器主动向客户端推送消息。本文将详细介绍在SpringBoot中实现网页消息推送的5种主要方法，每种方法都有其适用场景和优缺点。
1. Server-Sent Events (SSE)
Server-Sent Events是HTML5标准的一部分，允许服务器向客户端推送事件。它基于HTTP协议，实现简单，非常适合单向数据推送场景。
后端实现
@RestController
@RequestMapping("/api/sse")
publicclass SSEController {
/**
* 创建SSE连接端点
* @return SseEmitter对象，用于发送服务器推送事件
*/
@GetMapping("/connect")
public SseEmitter connect() {
// 创建SseEmitter对象，设置超时时间为30分钟
SseEmitter emitter = new SseEmitter(30 * 60 * 1000L);
try {
// 发送连接成功消息
emitter.send(SseEmitter.event()
.name("connect")
.data("SSE连接建立成功"));
} catch (IOException e) {
emitter.completeWithError(e);
}
// 模拟定时推送消息
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
executor.scheduleAtFixedRate(() -> {
try {
// 每5秒推送一次时间信息
emitter.send(SseEmitter.event()
.name("message")
.data("当前时间: " + LocalDateTime.now().toString()));
} catch (IOException e) {
emitter.complete();
executor.shutdown();
}
}, 0, 5, TimeUnit.SECONDS);
return emitter;
}
}
前端实现
// 创建EventSource对象连接SSE端点
const eventSource = new EventSource('/api/sse/connect');
// 监听连接事件
eventSource.addEventListener('connect', function(event) {
console.log('连接状态:', event.data);
});
// 监听消息事件
eventSource.addEventListener('message', function(event) {
console.log('收到消息:', event.data);
// 将消息显示在页面上
document.getElementById('messages').innerHTML += '<div>' + event.data + '</div>';
});
// 监听错误事件
eventSource.onerror = function(event) {
console.error('SSE连接错误:', event);
};
优点： 实现简单，浏览器原生支持，自动重连机制缺点： 只支持单向推送，连接数限制，不支持二进制数据
2. WebSocket
WebSocket提供了全双工通信能力，是实时交互应用的首选方案。SpringBoot通过Spring WebSocket提供了完善的支持。
配置WebSocket
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
// 注册WebSocket处理器，允许跨域访问
registry.addHandler(new MyWebSocketHandler(), "/websocket")
.setAllowedOrigins("*");
}
}
WebSocket处理器
@Component
publicclass MyWebSocketHandler extends TextWebSocketHandler {
// 存储所有活跃的WebSocket会话
privatestaticfinal Set<WebSocketSession> sessions =
Collections.synchronizedSet(new HashSet<>());
/**
* 连接建立后调用
*/
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
sessions.add(session);
System.out.println("WebSocket连接建立: " + session.getId());
// 向新连接的客户端发送欢迎消息
session.sendMessage(new TextMessage("欢迎连接WebSocket服务！"));
}
/**
* 接收到客户端消息时调用
*/
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message)
throws Exception {
String payload = message.getPayload();
System.out.println("收到消息: " + payload);
// 广播消息给所有连接的客户端
broadcastMessage("服务器回复: " + payload);
}
/**
* 连接关闭后调用
*/
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status)
throws Exception {
sessions.remove(session);
System.out.println("WebSocket连接关闭: " + session.getId());
}
/**
* 广播消息给所有连接的客户端
*/
public static void broadcastMessage(String message) {
synchronized (sessions) {
sessions.removeIf(session -> {
try {
if (session.isOpen()) {
session.sendMessage(new TextMessage(message));
returnfalse; // 保留该会话
}
} catch (Exception e) {
System.err.println("发送消息失败: " + e.getMessage());
}
returntrue; // 移除无效会话
});
}
}
}
前端实现
// 创建WebSocket连接
const socket = new WebSocket('ws://localhost:8080/websocket');
// 连接打开时的处理
socket.onopen = function(event) {
console.log('WebSocket连接已建立');
// 发送测试消息
socket.send('Hello WebSocket!');
};
// 接收消息的处理
socket.onmessage = function(event) {
console.log('收到消息:', event.data);
document.getElementById('messages').innerHTML += '<div>' + event.data + '</div>';
};
// 连接关闭时的处理
socket.onclose = function(event) {
console.log('WebSocket连接已关闭');
};
// 错误处理
socket.onerror = function(error) {
console.error('WebSocket错误:', error);
};
// 发送消息的函数
function sendMessage() {
const input = document.getElementById('messageInput');
if (socket.readyState === WebSocket.OPEN) {
socket.send(input.value);
input.value = '';
}
}
优点： 全双工通信，低延迟，支持二进制数据缺点： 实现相对复杂，需要处理连接状态管理
3. 长轮询 (Long Polling)
长轮询是一种模拟推送的技术，客户端发起请求后，服务器不立即响应，而是等待有数据时再返回。
后端实现
@RestController
@RequestMapping("/api/longpoll")
publicclass LongPollingController {
// 使用队列存储待推送的消息
privatefinal BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();
/**
* 长轮询端点
* @param timeout 超时时间（毫秒）
* @return 消息内容或超时提示
*/
@GetMapping("/poll")
public ResponseEntity<String> longPoll(
@RequestParam(defaultValue = "30000") long timeout) {
try {
// 等待消息，最多等待指定的超时时间
String message = messageQueue.poll(timeout, TimeUnit.MILLISECONDS);
if (message != null) {
// 有消息时返回消息内容
return ResponseEntity.ok(message);
} else {
// 超时时返回空响应
return ResponseEntity.ok("timeout");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body("服务器内部错误");
}
}
/**
* 发送消息到队列
* @param message 要发送的消息
* @return 发送结果
*/
@PostMapping("/send")
public ResponseEntity<String> sendMessage(@RequestBody String message) {
try {
messageQueue.offer(message);
return ResponseEntity.ok("消息发送成功");
} catch (Exception e) {
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body("消息发送失败");
}
}
}
前端实现
class LongPollingClient {
constructor() {
this.isPolling = false;
}
/**
* 开始长轮询
*/
startPolling() {
if (this.isPolling) return;
this.isPolling = true;
this.poll();
}
/**
* 停止长轮询
*/
stopPolling() {
this.isPolling = false;
}
/**
* 执行轮询请求
*/
async poll() {
while (this.isPolling) {
try {
const response = await fetch('/api/longpoll/poll?timeout=30000');
const message = await response.text();
if (message && message !== 'timeout') {
console.log('收到消息:', message);
this.handleMessage(message);
}
} catch (error) {
console.error('轮询错误:', error);
// 发生错误时等待一段时间再重试
awaitthis.sleep(5000);
}
}
}
/**
* 处理接收到的消息
*/
handleMessage(message) {
document.getElementById('messages').innerHTML +=
'<div>长轮询消息: ' + message + '</div>';
}
/**
* 延迟函数
*/
sleep(ms) {
returnnewPromise(resolve => setTimeout(resolve, ms));
}
}
// 使用示例
const client = new LongPollingClient();
client.startPolling();
优点： 实现简单，兼容性好，服务器控制推送时机缺点： 资源消耗大，延迟相对较高
4. 定时轮询 (Short Polling)
定时轮询是最简单的实现方式，客户端定期向服务器请求数据。
后端实现
@RestController
@RequestMapping("/api/shortpoll")
publicclass ShortPollingController {
// 模拟消息存储
privatefinal List<Message> messages = new ArrayList<>();
privatefinal AtomicLong messageIdCounter = new AtomicLong(0);
/**
* 获取指定ID之后的新消息
* @param lastMessageId 客户端最后接收的消息ID
* @return 新消息列表
*/
@GetMapping("/messages")
public ResponseEntity<List<Message>> getNewMessages(
@RequestParam(defaultValue = "0") Long lastMessageId) {
// 过滤出ID大于lastMessageId的消息
List<Message> newMessages = messages.stream()
.filter(msg -> msg.getId() > lastMessageId)
.collect(Collectors.toList());
return ResponseEntity.ok(newMessages);
}
/**
* 添加新消息
* @param content 消息内容
* @return 创建的消息
*/
@PostMapping("/messages")
public ResponseEntity<Message> addMessage(@RequestBody String content) {
Message message = new Message(
messageIdCounter.incrementAndGet(),
content,
LocalDateTime.now()
);
messages.add(message);
return ResponseEntity.ok(message);
}
/**
* 消息实体类
*/
publicstaticclass Message {
private Long id;
private String content;
private LocalDateTime timestamp;
public Message(Long id, String content, LocalDateTime timestamp) {
this.id = id;
this.content = content;
this.timestamp = timestamp;
}
// Getters and Setters
public Long getId() { return id; }
public void setId(Long id) { this.id = id; }
public String getContent() { return content; }
public void setContent(String content) { this.content = content; }
public LocalDateTime getTimestamp() { return timestamp; }
public void setTimestamp(LocalDateTime timestamp) { this.timestamp = timestamp; }
}
}
前端实现
class ShortPollingClient {
constructor(interval = 5000) {
this.interval = interval;
this.lastMessageId = 0;
this.isPolling = false;
this.timerId = null;
}
/**
* 开始定时轮询
*/
startPolling() {
if (this.isPolling) return;
this.isPolling = true;
this.poll(); // 立即执行一次
// 设置定时器
this.timerId = setInterval(() => {
this.poll();
}, this.interval);
}
/**
* 停止轮询
*/
stopPolling() {
this.isPolling = false;
if (this.timerId) {
clearInterval(this.timerId);
this.timerId = null;
}
}
/**
* 执行轮询请求
*/
async poll() {
if (!this.isPolling) return;
try {
const response = await fetch(
`/api/shortpoll/messages?lastMessageId=${this.lastMessageId}`
);
const messages = await response.json();
if (messages && messages.length > 0) {
messages.forEach(message => {
this.handleMessage(message);
this.lastMessageId = Math.max(this.lastMessageId, message.id);
});
}
} catch (error) {
console.error('轮询错误:', error);
}
}
/**
* 处理接收到的消息
*/
handleMessage(message) {
console.log('收到新消息:', message);
document.getElementById('messages').innerHTML +=
`<div>定时轮询消息[${message.id}]: ${message.content}</div>`;
}
}
// 使用示例
const client = new ShortPollingClient(3000); // 每3秒轮询一次
client.startPolling();
优点： 实现最简单，易于理解和调试缺点： 实时性差，资源浪费严重
5. Spring事件机制结合异步处理
利用Spring的事件机制可以实现解耦的消息推送系统，特别适合复杂的业务场景。
事件定义和处理
/**
* 消息推送事件
*/
publicclass MessagePushEvent extends ApplicationEvent {
privatefinal String message;
privatefinal String target; // 推送目标
public MessagePushEvent(Object source, String message, String target) {
super(source);
this.message = message;
this.target = target;
}
public String getMessage() { return message; }
public String getTarget() { return target; }
}
/**
* 事件监听器
*/
@Component
publicclass MessagePushEventListener {
privatefinal SimpMessagingTemplate messagingTemplate;
public MessagePushEventListener(SimpMessagingTemplate messagingTemplate) {
this.messagingTemplate = messagingTemplate;
}
/**
* 异步处理消息推送事件
*/
@EventListener
@Async
public void handleMessagePushEvent(MessagePushEvent event) {
try {
// 模拟业务处理延迟
Thread.sleep(100);
// 通过WebSocket推送消息
messagingTemplate.convertAndSend(
"/topic/messages/" + event.getTarget(),
event.getMessage()
);
System.out.println("消息推送完成: " + event.getMessage());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
/**
* 业务服务类
*/
@Service
publicclass MessageService {
privatefinal ApplicationEventPublisher eventPublisher;
public MessageService(ApplicationEventPublisher eventPublisher) {
this.eventPublisher = eventPublisher;
}
/**
* 发送消息
*/
public void sendMessage(String content, String target) {
// 执行业务逻辑...
// 发布推送事件
MessagePushEvent event = new MessagePushEvent(this, content, target);
eventPublisher.publishEvent(event);
}
}
WebSocket配置（用于最终推送）
@Configuration
@EnableWebSocketMessageBroker
publicclass WebSocketMessageConfig implements WebSocketMessageBrokerConfigurer {
/**
* 配置消息代理
*/
@Override
public void configureMessageBroker(MessageBrokerRegistry config) {
// 启用简单消息代理，并设置消息代理的目的地前缀
config.enableSimpleBroker("/topic", "/queue");
// 设置客户端向服务器发送消息的目的地前缀
config.setApplicationDestinationPrefixes("/app");
}
/**
* 注册STOMP端点
*/
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
// 注册STOMP端点，并允许跨域
registry.addEndpoint("/ws").setAllowedOriginPatterns("*").withSockJS();
}
}
优点： 解耦性强，易于扩展，支持复杂业务逻辑缺点： 配置相对复杂，需要结合其他推送技术使用
总结
每种消息推送方法都有其适用场景：
- SSE ：适合简单的单向数据推送，如实时通知、数据监控
- WebSocket ：适合需要双向交互的实时应用，如聊天室、在线游戏
- 长轮询 ：适合对实时性要求不太高但需要服务器主动推送的场景
- 短轮询 ：适合对实时性要求较低的简单场景
- Spring事件 ：适合复杂业务场景下的解耦式消息推送
在实际项目中，可以根据具体需求选择合适的方案，甚至可以组合使用多种方法来实现最佳的用户体验。重要的是要考虑服务器资源、网络环境、浏览器兼容性等因素，选择最适合的技术方案。