Server-Sent Events(SSE)是一种简单的技术,允许服务器向客户端推送实时更新。在Spring Boot项目中,我们可以使用SseEmitter类来实现SSE功能。本文将详细介绍如何在Spring Boot项目中使用SSE,并给出一个使用示例。
首先,客户端通过浏览器向服务器发送一个SSE请求,当服务器收到这个SSE请求后,会建立一个持久的HTTP连接,并将响应的Content-Type设置为text/event-stream。然后,服务器就可以通过这条已经建立的连接向客户端持续发送数据了。每个数据都由一个或多个字段组成,如event、data、id等。需要注意的是,SSE是单向通信协议,只能从服务器端向客户端发送数据。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
public class SseEmitterUtil {
private final static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
private final static AtomicInteger count = new AtomicInteger();
/**
* 创建SSE连接
* Connect sse emitter.
*
* @param clientId the client id
* @return the sse emitter
*/
public static SseEmitter connect(String clientId) {
try {
// 设置超时时间,0表示不过期。默认30秒
SseEmitter sseEmitter = new SseEmitter(0L);
// 注册回调
sseEmitter.onCompletion(() -> {
log.info("客户端正常关闭,clientId = {}", clientId);
removeClientId(clientId);
});
sseEmitter.onError(throwable -> {
log.error("连接出现异常,准备关闭,clientId = {}", clientId);
removeClientId(clientId);
});
sseEmitter.onTimeout(() -> {
log.info("连接已超时,准备关闭,clientId = {}", clientId);
removeClientId(clientId);
});
sseEmitterMap.put(clientId, sseEmitter);
count.getAndIncrement();
log.info("连接成功,客户端连接总数:{}", count.get());
return sseEmitter;
} catch (Exception e) {
log.info("创建新的sse连接异常,clientId:{}", clientId);
log.error(e.toString());
}
return null;
}
/**
* 给指定用户发送消息
* Send message.
*
* @param clientId the client id 客户端id
* @param message the message
*/
public static void sendMessage(String clientId, String message) {
if (sseEmitterMap.containsKey(clientId)) {
try {
sseEmitterMap.get(clientId).send(message);
} catch (IOException e) {
log.error("clientId:[{}],推送数据:{},推送异常:{}", clientId, message, e.getMessage());
removeClientId(clientId);
}
}
}
/**
* 断开连接
* Disconnect.
*
* @param clientId the client id
*/
public static void disconnect(String clientId) {
if (sseEmitterMap.containsKey(clientId)) {
log.info("连接断开,clientId:{}", clientId);
sseEmitterMap.get(clientId).complete();
} else {
log.error("不存在clientId为{}的客户端连接", clientId);
}
}
/**
* 移除连接客户端
* Remove client id.
*
* @param id the id 客户端连接ID
*/
private static void removeClientId(String id) {
sseEmitterMap.remove(id);
count.getAndDecrement();
log.info("剩余客户端连接数:{}", count.get());
}
}
@RequestMapping("/sse")
public class SseController {
/**
* 连接SSE
* Push sse emitter.
*
* @param clientId the client id
* @return the sse emitter
*/
@GetMapping("/connect")
public SseEmitter push(String clientId) {
return SseEmitterUtil.connect(clientId);
}
/**
* 消息推送
* Push.
*
* @param clientId the client id
* @param content the content
*/
@GetMapping("/push")
public void push(String clientId, String content) {
SseEmitterUtil.sendMessage(clientId, content);
}
/**
* 断开连接
* Disconnect.
*
* @param clientId the client id
*/
@GetMapping("/disconnect")
public void disconnect(String clientId) {
SseEmitterUtil.disconnect(clientId);
}
}
PS:在使用过程是前后端约定好不同的场景使用特定的clientId进行消息传输,使用完毕调用关闭连接的接口即可