diff --git a/yudao-module-iot/yudao-module-iot-biz/pom.xml b/yudao-module-iot/yudao-module-iot-biz/pom.xml index 8b26a5e..062a467 100644 --- a/yudao-module-iot/yudao-module-iot-biz/pom.xml +++ b/yudao-module-iot/yudao-module-iot-biz/pom.xml @@ -31,6 +31,12 @@ yudao-module-iot-core ${revision} + + + + + + cn.iocoder.boot diff --git a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java index 1fc608b..2327a32 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java +++ b/yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java @@ -8,10 +8,10 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.context.ApplicationContext; import org.springframework.context.event.EventListener; -import java.util.ArrayList; -import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; /** * 本地的 {@link IotMessageBus} 实现类 @@ -30,7 +30,7 @@ public class IotLocalMessageBus implements IotMessageBus { * 订阅者映射表 * Key: topic */ - private final Map>> subscribers = new HashMap<>(); + private final Map>> subscribers = new ConcurrentHashMap<>(); @Override public void post(String topic, Object message) { @@ -40,7 +40,7 @@ public class IotLocalMessageBus implements IotMessageBus { @Override public void register(IotMessageSubscriber subscriber) { String topic = subscriber.getTopic(); - List> topicSubscribers = subscribers.computeIfAbsent(topic, k -> new ArrayList<>()); + List> topicSubscribers = subscribers.computeIfAbsent(topic, k -> new CopyOnWriteArrayList<>()); topicSubscribers.add(subscriber); log.info("[register][topic({}/{}) 注册消费者({})成功]", topic, subscriber.getGroup(), subscriber.getClass().getName()); diff --git a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java index b282bc8..9e8b2d1 100644 --- a/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java +++ b/yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java @@ -10,7 +10,12 @@ import org.springframework.boot.test.context.SpringBootTest; import org.springframework.context.annotation.Import; import org.springframework.test.context.TestPropertySource; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -174,4 +179,69 @@ public class LocalIotMessageBusIntegrationTest { log.info("[测试] 多主题测试完成"); } + /** + * 1 topic 16 subscriber 并发注册 + */ + @Test + public void testConcurrentRegister() throws Exception { + // 准备参数 + String topic = "concurrent-register-topic"; + String testMessage = "concurrent"; + int threadCount = 16; + List receivedCounts = new ArrayList<>(threadCount); + List> subscribers = new ArrayList<>(threadCount); + for (int i = 0; i < threadCount; i++) { + AtomicInteger receivedCount = new AtomicInteger(0); + String group = "concurrent-group-" + i; + receivedCounts.add(receivedCount); + subscribers.add(new IotMessageSubscriber<>() { + + @Override + public String getTopic() { + return topic; + } + + @Override + public String getGroup() { + return group; + } + + @Override + public void onMessage(String message) { + assertEquals(testMessage, message); + receivedCount.incrementAndGet(); + } + + }); + } + + // 调用:16 个线程同时注册 + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + try { + CountDownLatch startLatch = new CountDownLatch(1); + List> futures = new ArrayList<>(threadCount); + for (IotMessageSubscriber subscriber : subscribers) { + futures.add(executor.submit(() -> { + startLatch.await(); + messageBus.register(subscriber); + return null; + })); + } + startLatch.countDown(); + // 注册时抛出的异常(例如 ConcurrentModificationException),会在 get 时抛出 + for (Future future : futures) { + future.get(10, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + // 调用:注册完成后,发送一次消息(本地总线同步投递) + messageBus.post(topic, testMessage); + + // 断言:每个订阅者都收到 1 条消息,没有因并发注册而丢失 + for (int i = 0; i < threadCount; i++) { + assertEquals(1, receivedCounts.get(i).get(), "订阅者 " + i + " 应该收到 1 条消息"); + } + } + } \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/pom.xml b/yudao-module-iot/yudao-module-iot-gateway/pom.xml index 2c2320b..c344606 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/pom.xml +++ b/yudao-module-iot/yudao-module-iot-gateway/pom.xml @@ -77,7 +77,7 @@ ${project.artifactId} - + org.springframework.boot spring-boot-maven-plugin @@ -87,6 +87,10 @@ repackage + + + exec + diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/IotDeviceServiceImpl.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/IotDeviceServiceImpl.java index fee48d1..00f32d0 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/IotDeviceServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/IotDeviceServiceImpl.java @@ -21,7 +21,7 @@ import static cn.iocoder.yudao.framework.common.util.cache.CacheUtils.buildAsync * * @author 芋道源码 */ -@Service +@Service("iotGatewayDeviceServiceImpl") // 显式 bean 名:避免与 biz 的 IotDeviceServiceImpl 同名冲突 @Slf4j public class IotDeviceServiceImpl implements IotDeviceService { diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java index 9b7818e..d616d11 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java @@ -43,7 +43,7 @@ import static cn.iocoder.yudao.module.iot.gateway.enums.ErrorCodeConstants.*; * * @author 芋道源码 */ -@Service +@Service("iotGatewayDeviceMessageServiceImpl") // 显式 bean 名:避免与 biz 的 IotDeviceMessageServiceImpl 同名冲突 @Slf4j public class IotDeviceMessageServiceImpl implements IotDeviceMessageService { diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/remote/IotDeviceApiImpl.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/remote/IotDeviceApiImpl.java index db820d2..30bf181 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/remote/IotDeviceApiImpl.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/remote/IotDeviceApiImpl.java @@ -15,10 +15,10 @@ import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterRespDTO; import cn.iocoder.yudao.module.iot.core.topic.auth.IotSubDeviceRegisterRespDTO; import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties; -import java.util.List; import jakarta.annotation.PostConstruct; import jakarta.annotation.Resource; import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.web.client.RestTemplateBuilder; import org.springframework.core.ParameterizedTypeReference; import org.springframework.http.HttpEntity; @@ -35,9 +35,12 @@ import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeC /** * Iot 设备信息 Service 实现类:调用远程的 device http 接口,进行设备认证、设备获取等 * + * 注意:仅配置 yudao.iot.gateway.rpc.url 时创建(独立部署);合并部署时,使用 biz 的 IotDeviceCommonApi 实现 + * * @author 芋道源码 */ @Service +@ConditionalOnProperty(prefix = "yudao.iot.gateway.rpc", name = "url") @Slf4j public class IotDeviceApiImpl implements IotDeviceCommonApi {