springboot整合SSE技术开发经验总结及心得
一、开发背景
公司需要开发一个大屏界面,大屏页面的数据是实时更新的,由后端主动实时推送数据给大屏页面。此时会立刻联想到:websocket 技术。当然使用websocket,确实可以解决这个场景。但是今天本文的主角是 :SSE,他和websocket略有不同,SSE只能由服务端主动发消息,而websocket前后端都可以推送消息。
二、快速了解SSE
1、概念
SSE全称 Server Sent Event,顾名思义,就是服务器发送事件,所以也就注定了他 只能由服务端发送信息。
2、特性
三、开发思路
四、代码演示
1、引入依赖
原则上是不需要引入的,因为springboot底层已经整合了SSE
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
2、服务端代码
@RestController
@CrossOrigin
@RequestMapping("/sse")
public class SseEmitterController extends BaseController {
@Autowired
private SseEmitterService sseEmitterService;
/**
* 创建SSE连接
*
* @return
*/
@GetMapping("/connect/{type}")
public SseEmitter connect(@PathVariable("type") String type) {
return sseEmitterService.connect(type);
}
}
public interface SseEmitterService {
SseEmitter connect(String type);
void volumeOverview();
void sysOperation();
void monitor();
........
}
@Service
public class SseEmitterServiceImpl implements SseEmitterService {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private static Map<String, SseEmitterUTF8> sseCache = new ConcurrentHashMap<>();
/**
* 创建连接sse
* @param type
* @return
*/
@Override
public SseEmitter connect(String type) {
final String clientId = UUID.randomUUID().toString().replace("-", "");
SseEmitterUTF8 sseEmitter = new SseEmitterUTF8(0L);
try {
sseEmitter.send(SseEmitter.event().comment("创建连接成功 !!!"));
} catch (IOException e) {
logger.error("创建连接失败 , {} " , e.getMessage());
}
sseEmitter.onCompletion(() -> {
logger.info("connect onCompletion , {} 结束连接 ..." , clientId);
removeClient(clientId);
});
sseEmitter.onTimeout(() -> {
logger.info("connect onTimeout , {} 连接超时 ..." , clientId);
removeClient(clientId);
});
sseEmitter.onError((throwable) -> {
logger.error("connect onError , {} 连接异常 ..." , clientId);
removeClient(clientId);
});
sseCache.put(clientId, sseEmitter);
//立即推送
volumeOverview();
dealResp();
monitor();
if (type.equals(SseEmitterConstant.OVER_VIEW)){
sysOperation();
mileStone();
}
logger.info("当前用户总连接数 : {} " , sseCache.size());
return sseEmitter;
}
/**
* 交易量概览
*/
@Override
public void volumeOverview() {
Map<String,Object> map = new HashMap<>();
map.put("latest_tps",440.3);
map.put("total_cics_trans",341656001);
map.put("total_zjcx_trans",391656001);
map.put("zjcx_tps",23657);
map.put("day10",48388352);
map.put("history",105013985);
SseEmitter.SseEventBuilder data = SseEmitter.event().name(SseEmitterConstant.VOLUME_OVERVIEW).data(map, MediaType.APPLICATION_JSON);
for (Map.Entry<String, SseEmitterUTF8> entry : sseCache.entrySet()) {
SseEmitterUTF8 sseEmitter = entry.getValue();
if (sseEmitter == null) {
continue;
}
try {
sseEmitter.send(data);
} catch (IOException e) {
String body = "SseEmitterServiceImpl[volumeOverview ]";
logger.error(body + ": 向客户端 {} 推送消息失败 , 尝试进行重推 : {}", entry.getKey() ,e.getMessage());
messageRepush(entry.getKey(),data,body);
}
}
}
private void messageRepush(String type, SseEmitter.SseEventBuilder data,String body){
for (int i = 0; i < 3; i++) {
try {
Thread.sleep(2000);
SseEmitterUTF8 sseEmitter = sseCache.get(type);
if (sseEmitter == null) {
logger.error(body + " :向客户端{} 第{}次消息重推失败,未创建长链接", type, i + 1);
continue;
}
sseEmitter.send(data);
} catch (Exception ex) {
logger.error(body + " :向客户端{} 第{}次消息重推失败", type, i + 1, ex);
continue;
}
logger.info(body + " :向客户端{} 第{}次消息重推成功", type, i + 1);
return;
}
}
常量类
public class SseEmitterConstant {
/**
* 创建连接的客户端类型
*/
public static final String OVER_VIEW = "overview";
/**
* even 数据类型
*/
public static final String VOLUME_OVERVIEW = "vw";
public SseEmitterConstant(){}
}
3、后端定时任务代码
采用注解的方式实现:@Scheduled,使用该注解时,需要增加这个注解@EnableScheduling,相当于来开启定时调度功能,如果不加@EnableScheduling注解,那么定时调度会不生效的。
package com.hidata;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.scheduling.annotation.EnableScheduling;
@SpringBootApplication(exclude = { DataSourceAutoConfiguration.class })
@EnableScheduling
public class HidataApplication {
public static void main(String[] args)
{
SpringApplication.run(HidataApplication.class, args);
System.out.println("[HiUrlShorter platform startup!]");
}
}
创建 定时任务调度类,在该类上加上@Scheduled注解,
@Configuration
public class SendMessageTask{
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private SseEmitterService sseEmitterService;
@Scheduled(cron = "0/40 * * * * ?}")
public void volumeOverviewTask() {
try {
sseEmitterService.volumeOverview();
} catch (Exception e) {
logger.error("SendMessageTask [volumeOverviewTask]: {} ",e.getMessage());
}
}
.......
}
4、解决乱码的实体类
如果发送中文数据的时候,会出现乱码的现象。此时需要做对应的处理
package com.hidata.devops.lagrescreen.domain;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.nio.charset.StandardCharsets;
public class SseEmitterUTF8 extends SseEmitter {
public SseEmitterUTF8(Long timeout) {
super(timeout);
}
@Override
protected void extendResponse(ServerHttpResponse outputMessage) {
super.extendResponse(outputMessage);
HttpHeaders headers = outputMessage.getHeaders();
headers.setContentType(new MediaType(MediaType.TEXT_EVENT_STREAM, StandardCharsets.UTF_8));
}
}
4、前端代码
// 连接服务器
var sseSource = new EventSource("http://localhost:8080/sse/connect");
// 连接打开
sseSource.onopen = function () {
console.log("连接打开");
}
// 连接错误
sseSource.onerror = function (err) {
console.log("连接错误:", err);
}
//接收信息
eventSource.addEventListener("vw", function (event) {
console.log(event.data);
.....
});
五、核心代码分析
先看代码片段
SseEmitter.event().name("vw").data(map, MediaType.APPLICATION_JSON);
分析:
后端不会把所有数据一起发送给前端,而是会把页面分成多个模块,然后发给前端,此时前端需要区分哪一块数据对应哪一块页面。所以我们可以给各个模块的数据起个名字。也就是上述的代码
SseEmitter.event().name("vw")
这样,前端就知道怎么渲染页面了,类似于这样
关于even()的属性,可以查看源码,
public interface SseEventBuilder {
SseEmitter.SseEventBuilder id(String var1);
SseEmitter.SseEventBuilder name(String var1);
SseEmitter.SseEventBuilder reconnectTime(long var1);
SseEmitter.SseEventBuilder comment(String var1);
SseEmitter.SseEventBuilder data(Object var1);
SseEmitter.SseEventBuilder data(Object var1, @Nullable MediaType var2);
Set<DataWithMediaType> build();
}
原文地址:https://blog.csdn.net/weixin_43860634/article/details/134378074
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。
如若转载,请注明出处:http://www.7code.cn/show_10065.html
如若内容造成侵权/违法违规/事实不符,请联系代码007邮箱:suwngjj01@126.com进行投诉反馈,一经查实,立即删除!