优化

liusuyi

2026-02-10 4ac9edd4c0030d2daafd712cd14ef9080b2f1e43

 ard-sample/src/main/java/com/ard/sample/component/websocket/handler/DockWebSocketHandler.java

@@ -4,7 +4,6 @@
import com.ard.sample.component.websocket.enums.MessageType;
import com.ard.sample.component.websocket.model.Payload;
import com.ard.sample.component.websocket.model.WebSocketMessage;
import com.ard.sample.manage.service.IControlService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -31,9 +30,6 @@

    @Autowired
    private LiveStreamHandler _liveStreamService;

    @Autowired
    private IControlService _controlService;

    // ====================== 心跳相关常量（可配置化建议放到配置文件） ======================
    /** 心跳间隔（秒）：客户端需每隔此时间发送一次PING消息 */
@@ -153,6 +149,13 @@
            // 3. 处理订阅/取消订阅逻辑
            handleSubscription(userId, messageType, payload);

            // 直播类型订阅：触发开启直播
            if (MessageType.LIVESTREAM.getType().equals(messageType) && payload.isSubscribe()) {
                String sn = payload.getSn();
                _liveStreamService.StartLiveStreaming(sn,userId);
                log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
            }

        } catch (Exception e) {
            log.error("用户[{}]消息处理异常", userId, e);
            sendMessage(session, "消息格式错误或处理失败：" + e.getMessage());
@@ -187,11 +190,11 @@
            subscriptions.computeIfAbsent(subscriptionKey, k -> ConcurrentHashMap.newKeySet()).add(userId);
//            log.info("用户[{}]订阅设备[{}]的[{}]类型消息", userId, sn, type);

            // 直播类型订阅：触发开启直播
            if (MessageType.LIVESTREAM.getType().equals(type)) {
                _liveStreamService.StartLiveStreaming(sn);
                log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
            }
//            // 直播类型订阅：触发开启直播
//            if (MessageType.LIVESTREAM.getType().equals(type)) {
//                _liveStreamService.StartLiveStreaming(sn);
//                log.info("设备[{}]直播流已开启（用户[{}]订阅触发）", sn, userId);
//            }
        } else {
            // 取消订阅操作
            if (subscriptions.containsKey(subscriptionKey)) {
@@ -257,7 +260,7 @@
     * @param type    消息类型（对应MessageType）
     * @param content 消息内容（JSON字符串）
     */
    public void sendMessageToSubscribers(String sn, String type, String content) {
    public void sendMessageToSubscribers(String sn, String type, String content, String uId) {
        String subscriptionKey = getSubscriptionKey(sn, type);
        Set<String> subscriberIds = subscriptions.getOrDefault(subscriptionKey, Collections.emptySet());

@@ -280,6 +283,12 @@

            // 线程安全发送消息
            try {
                if(MessageType.LIVESTREAM.getType().equals(type)) {
                    WebSocketSession liveSession = userSessions.get(uId);
                    sendMessage(liveSession, content);
                    log.info("单独向用户[{}]推送设备[{}]的[{}]类型消息：{}", uId, sn, type, content);
                    return;
                }
                sendMessage(session, content);
                log.debug("向用户[{}]推送设备[{}]的[{}]类型消息：{}", userId, sn, type, content);
            } catch (IOException e) {