diff --git a/yudao-dependencies/pom.xml b/yudao-dependencies/pom.xml
index 3ff1534..ae72b99 100644
--- a/yudao-dependencies/pom.xml
+++ b/yudao-dependencies/pom.xml
@@ -76,7 +76,6 @@
- * 用于保证 WebSocket 重连操作的线程安全,避免多线程同时重连导致的资源竞争 - * - * @author HUIHUI - */ -@Repository -public class IotWebSocketLockRedisDAO { - - /** - * 锁等待超时时间(毫秒) - */ - private static final long LOCK_WAIT_TIME_MS = 5000; - - /** - * 锁持有超时时间(毫秒) - */ - private static final long LOCK_LEASE_TIME_MS = 10000; - - @Resource - private RedissonClient redissonClient; - - /** - * 在分布式锁保护下执行操作 - * - * @param serverUrl WebSocket 服务器地址 - * @param runnable 需要执行的操作 - * @throws Exception 如果获取锁超时或执行操作时发生异常 - */ - public void lock(String serverUrl, Runnable runnable) throws Exception { - String lockKey = formatKey(serverUrl); - RLock lock = redissonClient.getLock(lockKey); - - try { - // 尝试获取分布式锁 - boolean acquired = lock.tryLock(LOCK_WAIT_TIME_MS, LOCK_LEASE_TIME_MS, TimeUnit.MILLISECONDS); - if (!acquired) { - throw new RuntimeException("获取 WebSocket 连接锁超时,服务器: " + serverUrl); - } - - // 执行操作 - runnable.run(); - } finally { - // 释放锁 - if (lock.isHeldByCurrentThread()) { - lock.unlock(); - } - } - } - - private static String formatKey(String serverUrl) { - return String.format(WEBSOCKET_CONNECT_LOCK, serverUrl); - } - -} diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java index 6515629..ebfe1f8 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/data/action/IotWebSocketDataRuleAction.java @@ -3,13 +3,15 @@ package cn.iocoder.yudao.module.iot.service.rule.data.action; import cn.hutool.core.util.StrUtil; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.dal.dataobject.rule.config.IotDataSinkWebSocketConfig; -import cn.iocoder.yudao.module.iot.dal.redis.rule.IotWebSocketLockRedisDAO; import cn.iocoder.yudao.module.iot.enums.rule.IotDataSinkTypeEnum; import cn.iocoder.yudao.module.iot.service.rule.data.action.websocket.IotWebSocketClient; -import jakarta.annotation.Resource; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; + /** * WebSocket 的 {@link IotDataRuleAction} 实现类 *
@@ -24,8 +26,17 @@ import org.springframework.stereotype.Component;
public class IotWebSocketDataRuleAction extends
IotDataRuleCacheableAction
+ * WebSocket 连接是与特定服务器实例绑定的,使用单机锁即可保证重连的线程安全
+ */
+ private final ConcurrentHashMap