优化报警推送心跳定时任务
parent
caaaaacb89
commit
ace7b78498
|
@ -1,6 +1,5 @@
|
||||||
package com.genersoft.iot.vmp.gb28181.controller;
|
package com.genersoft.iot.vmp.gb28181.controller;
|
||||||
|
|
||||||
import com.genersoft.iot.vmp.conf.DynamicTask;
|
|
||||||
import com.genersoft.iot.vmp.gb28181.event.alarm.AlarmEventListener;
|
import com.genersoft.iot.vmp.gb28181.event.alarm.AlarmEventListener;
|
||||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||||
import org.springframework.web.bind.annotation.GetMapping;
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
@ -29,9 +28,6 @@ public class SseController {
|
||||||
@Resource
|
@Resource
|
||||||
private AlarmEventListener alarmEventListener;
|
private AlarmEventListener alarmEventListener;
|
||||||
|
|
||||||
@Resource
|
|
||||||
private DynamicTask dynamicTask;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* SSE 推送.
|
* SSE 推送.
|
||||||
*
|
*
|
||||||
|
@ -45,17 +41,7 @@ public class SseController {
|
||||||
public void emit(HttpServletResponse response, @RequestParam String browserId) throws IOException, InterruptedException {
|
public void emit(HttpServletResponse response, @RequestParam String browserId) throws IOException, InterruptedException {
|
||||||
response.setContentType("text/event-stream");
|
response.setContentType("text/event-stream");
|
||||||
response.setCharacterEncoding("utf-8");
|
response.setCharacterEncoding("utf-8");
|
||||||
|
|
||||||
PrintWriter writer = response.getWriter();
|
PrintWriter writer = response.getWriter();
|
||||||
alarmEventListener.addSseEmitter(browserId, writer);
|
alarmEventListener.addSseEmitter(browserId, writer);
|
||||||
|
|
||||||
dynamicTask.startCron("sse-key", new Runnable() {
|
|
||||||
@Override
|
|
||||||
public void run() {
|
|
||||||
writer.write(":keep alive\n\n");
|
|
||||||
writer.flush();
|
|
||||||
}
|
|
||||||
}, 1000);
|
|
||||||
alarmEventListener.removeSseEmitter(browserId, writer);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -22,16 +22,27 @@ import java.util.concurrent.ConcurrentHashMap;
|
||||||
@Component
|
@Component
|
||||||
public class AlarmEventListener implements ApplicationListener<AlarmEvent> {
|
public class AlarmEventListener implements ApplicationListener<AlarmEvent> {
|
||||||
|
|
||||||
private static final Map<String, PrintWriter> SSE_CACHE = new ConcurrentHashMap<>();
|
private static final Map<String, PrintWriter> sseChannelMap = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
|
public void addSseEmitter(String browserId, PrintWriter writer) throws InterruptedException {
|
||||||
|
sseChannelMap.put(browserId, writer);
|
||||||
|
log.info("[SSE推送] 连接已建立, 浏览器 ID: {}, 当前在线数: {}", browserId, sseChannelMap.size());
|
||||||
|
while (!writer.checkError()) {
|
||||||
|
try {
|
||||||
|
Thread.sleep(1000);
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
throw new RuntimeException(e);
|
||||||
|
}
|
||||||
|
writer.write(":keep alive\n\n");
|
||||||
|
writer.flush();
|
||||||
|
}
|
||||||
|
removeSseEmitter(browserId, writer);
|
||||||
|
|
||||||
public void addSseEmitter(String browserId, PrintWriter writer) {
|
|
||||||
SSE_CACHE.put(browserId, writer);
|
|
||||||
log.info("[SSE推送] 连接已建立, 浏览器 ID: {}, 当前在线数: {}", browserId, SSE_CACHE.size());
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public void removeSseEmitter(String browserId, PrintWriter writer) {
|
public void removeSseEmitter(String browserId, PrintWriter writer) {
|
||||||
SSE_CACHE.remove(browserId, writer);
|
sseChannelMap.remove(browserId, writer);
|
||||||
log.info("[SSE推送] 连接已断开, 浏览器 ID: {}, 当前在线数: {}", browserId, SSE_CACHE.size());
|
log.info("[SSE推送] 连接已断开, 浏览器 ID: {}, 当前在线数: {}", browserId, sseChannelMap.size());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -42,13 +53,7 @@ public class AlarmEventListener implements ApplicationListener<AlarmEvent> {
|
||||||
|
|
||||||
log.info("设备报警事件触发, deviceId: {}, {}", event.getAlarmInfo().getDeviceId(), event.getAlarmInfo().getAlarmDescription());
|
log.info("设备报警事件触发, deviceId: {}, {}", event.getAlarmInfo().getDeviceId(), event.getAlarmInfo().getAlarmDescription());
|
||||||
|
|
||||||
|
for (Iterator<Map.Entry<String, PrintWriter>> it = sseChannelMap.entrySet().iterator(); it.hasNext(); ) {
|
||||||
String msg = "<strong>设备:</strong> <i>" + event.getAlarmInfo().getDeviceId() + "</i>"
|
|
||||||
+ "<br><strong>通道编号:</strong> <i>" + event.getAlarmInfo().getChannelId() + "</i>"
|
|
||||||
+ "<br><strong>报警描述:</strong> <i>" + event.getAlarmInfo().getAlarmDescription() + "</i>"
|
|
||||||
+ "<br><strong>报警时间:</strong> <i>" + event.getAlarmInfo().getAlarmTime() + "</i>";
|
|
||||||
|
|
||||||
for (Iterator<Map.Entry<String, PrintWriter>> it = SSE_CACHE.entrySet().iterator(); it.hasNext(); ) {
|
|
||||||
Map.Entry<String, PrintWriter> response = it.next();
|
Map.Entry<String, PrintWriter> response = it.next();
|
||||||
|
|
||||||
try {
|
try {
|
||||||
|
@ -59,12 +64,6 @@ public class AlarmEventListener implements ApplicationListener<AlarmEvent> {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
String sseMsg = "event:message\n" +
|
|
||||||
"data:" + msg + "\n" +
|
|
||||||
"\n";
|
|
||||||
System.out.println(
|
|
||||||
SSEMessage.getInstance("message", event.getAlarmInfo()).ecode()
|
|
||||||
);
|
|
||||||
writer.write(SSEMessage.getInstance("message", event.getAlarmInfo()).ecode());
|
writer.write(SSEMessage.getInstance("message", event.getAlarmInfo()).ecode());
|
||||||
writer.flush();
|
writer.flush();
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
|
|
Loading…
Reference in New Issue