优化日志打印

liusuyi

2025-10-20 dea4343d9f92a44e45eaeaab0afad6ede9ae5bb9

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
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.debug("给用户{}定时推送",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();
}