feat(iot): 支持设备网关与业务服务合并部署

- 网关提供普通依赖包，保留 exec 独立启动包 - 区分网关与业务 Bean，按配置启用远程设备 API - 修复本地消息总线并发注册并补充回归用例

feat(iot): 支持设备网关与业务服务合并部署
- 网关提供普通依赖包，保留 exec 独立启动包 - 区分网关与业务 Bean，按配置启用远程设备 API - 修复本地消息总线并发注册并补充回归用例
YunaiV
1 parent 37760883
Showing 7 changed files with 91 additions and 8 deletions
yudao-module-iot/yudao-module-iot-biz/pom.xml
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java
yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java
yudao-module-iot/yudao-module-iot-gateway/pom.xml
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/IotDeviceServiceImpl.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/message/IotDeviceMessageServiceImpl.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/service/device/remote/IotDeviceApiImpl.java
@@ -31,6 +31,12 @@
             <artifactId>yudao-module-iot-core</artifactId>
             <version>${revision}</version>
         </dependency>
+        <!-- 可选：与 biz 合并部署时引入 gateway，并将网关配置合并至 yudao-server 的 application.yaml，消息总线配置为 local。参见 https://t.zsxq.com/FtbO6 -->
+<!--        <dependency>-->
+<!--            <groupId>cn.iocoder.boot</groupId>-->
+<!--            <artifactId>yudao-module-iot-gateway</artifactId>-->
+<!--            <version>${revision}</version>-->
+<!--        </dependency>-->
         <dependency>
             <groupId>cn.iocoder.boot</groupId>
@@ -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<String, List<IotMessageSubscriber<?>>> subscribers = new HashMap<>();
+    private final Map<String, List<IotMessageSubscriber<?>>> 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<IotMessageSubscriber<?>> topicSubscribers = subscribers.computeIfAbsent(topic, k -> new ArrayList<>());
+        List<IotMessageSubscriber<?>> topicSubscribers = subscribers.computeIfAbsent(topic, k -> new CopyOnWriteArrayList<>());
         topicSubscribers.add(subscriber);
         log.info("[register][topic({}/{}) 注册消费者({})成功]",
                 topic, subscriber.getGroup(), subscriber.getClass().getName());
@@ -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<AtomicInteger> receivedCounts = new ArrayList<>(threadCount);
+        List<IotMessageSubscriber<String>> 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<Future<?>> futures = new ArrayList<>(threadCount);
+            for (IotMessageSubscriber<String> 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
@@ -77,7 +77,7 @@
         <!-- 设置构建的 jar 包名 -->
         <finalName>${project.artifactId}</finalName>
         <plugins>
-            <!-- 打包 -->
+            <!-- 打包：普通 jar 供 biz 引入，-exec.jar 独立运行 -->
             <plugin>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
@@ -87,6 +87,10 @@
                         <goals>
                             <goal>repackage</goal> <!-- 将引入的 jar 打入其中 -->
                         </goals>
+                        <configuration>
+                            <!-- 可执行 jar 使用 exec 分类，主 artifact 保留为普通可依赖 jar -->
+                            <classifier>exec</classifier>
+                        </configuration>
                     </execution>
                 </executions>
             </plugin>
@@ -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 {
@@ -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 {
@@ -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 {