package com.ard.work.component.websocket.handler;
|
|
import com.alibaba.fastjson2.JSON;
|
import com.ard.work.domain.message.BaseMessage;
|
import com.ard.work.component.websocket.utils.WebSocketUtil;
|
import lombok.extern.slf4j.Slf4j;
|
|
import java.util.HashSet;
|
import java.util.Map;
|
import java.util.Set;
|
import java.util.concurrent.*;
|
|
@Slf4j
|
public abstract class BaseSubscriptionManager<T> {
|
|
// 存储订阅的用户
|
protected final Map<String, Set<Long>> subscriptions = new ConcurrentHashMap<>();
|
// 定时任务调度器
|
protected final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
|
// 存储推送任务
|
protected final Map<String, ScheduledFuture<?>> pushTasks = new ConcurrentHashMap<>();
|
|
/**
|
* 订阅指定设备的数据
|
*/
|
public void subscribe(String boxSn, Long userId,Integer period) {
|
subscriptions.computeIfAbsent(boxSn, k -> new HashSet<>()).add(userId);
|
if (subscriptions.get(boxSn).size() == 1) {
|
startPushTask(boxSn,period);
|
}
|
}
|
|
/**
|
* 取消订阅
|
*/
|
public void unsubscribe(String boxSn, Long userId) {
|
Set<Long> userIds = subscriptions.get(boxSn);
|
if (userIds != null) {
|
userIds.remove(userId);
|
if (userIds.isEmpty()) {
|
stopPushTask(boxSn);
|
}
|
}
|
}
|
|
/**
|
* 启动推送任务
|
*/
|
private void startPushTask(String boxSn,Integer period) {
|
// 如果已经存在任务，则不再创建
|
if (pushTasks.containsKey(boxSn)) {
|
return;
|
}
|
ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
|
sendDataToSubscribers(boxSn);
|
}, 0, period, TimeUnit.SECONDS);
|
|
pushTasks.put(boxSn, future);
|
}
|
|
/**
|
* 停止推送任务
|
*/
|
private void stopPushTask(String boxSn) {
|
ScheduledFuture<?> future = pushTasks.get(boxSn);
|
if (future != null) {
|
future.cancel(false);
|
}
|
pushTasks.remove(boxSn);
|
}
|
|
/**
|
* 移除用户订阅
|
*/
|
public void removeSubscribe(Long userId) {
|
for (Set<Long> userIds : subscriptions.values()) {
|
userIds.remove(userId);
|
}
|
}
|
|
|
|
/**
|
* 向订阅者发送数据
|
*/
|
private void sendDataToSubscribers(String boxSn) {
|
T data = getData(boxSn);
|
BaseMessage<T> baseMessage = new BaseMessage<>(getMessageType(), data);
|
Set<Long> userIds = subscriptions.get(boxSn);
|
if (userIds != null) {
|
for (Long userId : userIds) {
|
try {
|
log.warn("给用户{}定时推送",userId);
|
WebSocketUtil.sendMessage(userId, JSON.toJSONString(baseMessage));
|
} catch (Exception e) {
|
log.error("Error sending message to session {}", userId, e);
|
}
|
}
|
}
|
}
|
|
/**
|
* 获取数据的具体实现
|
*/
|
protected abstract T getData(String boxSn);
|
|
/**
|
* 获取消息类型
|
*/
|
protected abstract String getMessageType();
|
}
|
