Spring MVC 通过 SSE 实现消息推送

又好久没有写文章了，自从有了大模型之后写文章的态度越来越提不起兴趣了，有问题，直接问大模型即可。前几天公司有个需求，想用 SSE 实现，之前从没写过，所以让大模型直接写，然后实现超级简单：

1. 编写 SSE 服务，来进行创建链接和发送消息

package cn.bridgeli.demo;

import lombok.Getter;  
import lombok.extern.slf4j.Slf4j;  
import org.apache.commons.collections4.CollectionUtils;  
import org.springframework.stereotype.Service;  
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;  
import java.util.List;  
import java.util.Map;  
import java.util.concurrent.ConcurrentHashMap;

@Slf4j  
@Getter  
@Service  
public class SseService {

private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

public SseEmitter stream(String usrId) {  
SseEmitter emitter = emitters.computeIfAbsent(usrId, k -> new SseEmitter(Long.MAX_VALUE));

emitter.onCompletion(() -> {  
log.info("SSE emitter completed");  
emitters.remove(usrId);  
});

emitter.onError((throwable) -> {  
log.error("Error occurred in SSE emitter", throwable);  
emitter.complete();  
emitters.remove(usrId);  
});

emitter.onTimeout(() -> {  
log.warn("SSE emitter timed out");  
emitter.complete();  
emitters.remove(usrId);  
});  
// 可选：连接成功时向客户端发送一个初始事件  
try {  
emitter.send(SseEmitter.event().name("connect").data("连接成功"));  
} catch (IOException e) {  
log.error("Error occurred while sending initial event", e);  
emitter.completeWithError(e);  
}

return emitter;  
}

public void send(List<String> userIds, String name, Object object) {  
if (!emitters.isEmpty()) {  
// 遍历所有用户的 SseEmitter，推送数据  
if (CollectionUtils.isEmpty(userIds)) {  
emitters.forEach((userId, emitter) -> {  
try {  
emitter.send(SseEmitter.event().name(name).data(object));  
} catch (IOException e) {  
// 如果发送失败，则移除该用户的 emitter  
log.error("Error occurred while sending event to user {}", userId, e);  
emitter.completeWithError(e);  
emitters.remove(userId);  
}  
});  
} else {  
userIds.forEach(userId -> {  
SseEmitter emitter = emitters.get(userId);  
if (emitter != null) {  
try {  
emitter.send(SseEmitter.event().name(name).data(object));  
} catch (IOException e) {  
// 如果发送失败，则移除该用户的 emitter  
log.error("Error occurred while sending event to user {}", userId, e);  
emitter.completeWithError(e);  
emitters.remove(userId);  
}  
}  
});  
}  
}  
}  
}

2. 编写对应的 Controller 给前端提供接口：

package cn.bridgeli.demo;

import cn.bridgeli.BaseAuthController;  
import io.swagger.v3.oas.annotations.tags.Tag;  
import jakarta.annotation.Resource;  
import lombok.extern.slf4j.Slf4j;  
import org.springframework.http.MediaType;  
import org.springframework.web.bind.annotation.GetMapping;  
import org.springframework.web.bind.annotation.RequestMapping;  
import org.springframework.web.bind.annotation.RestController;  
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@Slf4j  
@RestController  
@Tag(name = "SSE 推送服务")  
@RequestMapping("/auth/common/sse")  
public class SseController extends BaseAuthController {

@Resource  
private SseService sseService;

@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)  
public SseEmitter stream() {  
return sseService.stream(getLoginUsr().getUsrId());  
}  
}

3. 消息推送具体实现：

package cn.bridgeli.demo;

import cn.bridgeli.common.SseService;  
import cn.bridgeli.monitor.MonitorService;  
import cn.bridgeli.vo.CpuInfoVo;  
import jakarta.annotation.Resource;  
import lombok.extern.slf4j.Slf4j;  
import org.springframework.scheduling.annotation.Scheduled;  
import org.springframework.stereotype.Component;  
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.Map;

@Component  
@Slf4j  
public class ScheduledTask {

@Resource  
private MonitorService monitorService;  
@Resource  
private SseService sseService;

/**  
* 每分钟执行一次  
*/  
@Scheduled(cron = "0 0/1 \* \* * ?")  
public void updateOrderStatus() {  
log.info("=============定时任务=============");  
Map<String, SseEmitter> emitters = sseService.getEmitters();  
if (null == emitters || emitters.isEmpty()) {  
log.info("sse emitters is empty");  
return;  
}  
CpuInfoVo cpuData = monitorService.getCpuData();  
sseService.send(null, "cpu", cpuData);  
}

}

其实就是前端连接之后创建一个连接，保存连接，然后别的地方产生消息，推送消息，我的实例是通过 oshi 获取 CPU 使用率，实现对 CPU 的实时监控。