中州养老项目实战5:IoT智能监测系统

Walker0527

239人浏览 · 2026-08-20 09:16:19

Walker0527 · 2026-08-20 09:16:19 发布

中州养老项目实战

第五篇：IoT 智能监测系统详解

华为云 IoTDA · AMQP 消息处理 · 设备管理 · 线程池优化

一、IoT 核心概念

1.1 物联网四层架构

┌──────────────────────────────────┐
│                                应用层（业务系统）                 │
│                           数据展示 / 报警分析 / 报表                  │
├──────────────────────────────────┤
│                                平台层（IoT 平台）                       │
│     华为云IoTDA / 设备管理 / 规则引擎 │
├──────────────────────────────────┤
│                               网络层（通信协议）    │
│     MQTT / AMQP / CoAP / HTTP    │
├──────────────────────────────────┤
│                                感知层（硬件设备）                │
│                 智能手表 / 睡眠监测带 / 烟雾报警器              │
└──────────────────────────────────┘

1.2 核心概念表

概念	说明
产品	设备集合，定义同一类设备的功能（物模型）
物模型	描述设备功能和属性
设备	实际的物理设备，归属于某个产品
设备影子	存储设备最新上报数据和期望状态
MQTT	轻量级物联网通信协议，发布/订阅模式
AMQP	高级消息队列协议，服务端异步接收设备数据

二、华为云 IoTDA 平台配置

2.1 物模型定义

智能手表物模型示例：

{
  "service_id": "health_monitor",
  "properties": [
    {
      "property_name": "heart_rate",
      "data_type": "int",
      "description": "心率",
      "min": "0", "max": "300", "unit": "bpm"
    },
    {
      "property_name": "blood_oxygen",
      "data_type": "int",
      "description": "血氧饱和度",
      "min": "0", "max": "100", "unit": "%"
    },
    {
      "property_name": "body_temperature",
      "data_type": "float",
      "description": "体温",
      "min": "35.0", "max": "42.0", "unit": "℃"
    }
  ]
}

2.2 配置类

@Component
@ConfigurationProperties(prefix = "huawei.iot")
@Data
public class HuaWeiIotConfigProperties {
    private String host;
    private Integer port;
    private String accessKey;
    private String accessCode;
    private String queueName;
    private String vhost;
    private Integer idleTimeout;
    private String saslMechanisms;
    private Integer queuePrefetch;
}

三、设备管理

3.1 设备注册

设备注册需要同时写入 IoT 平台和本地数据库：

@Override
public void registerDevice(DeviceDto dto) {
    // 1. 校验设备名称唯一性
    // 2. 校验设备标识码唯一性
    // 3. 校验同一位置不重复绑定相同产品
   
    // 4. 调用华为云 SDK 注册设备
    AddDeviceRequest request = new AddDeviceRequest();
    AddDevice body = new AddDevice();
    body.withProductId(dto.getProductKey());
    body.withDeviceName(dto.getDeviceName());
    body.withNodeId(dto.getNodeId());
   
    AuthInfo authInfo = new AuthInfo();
    String secret = UUID.randomUUID().toString().replaceAll("-", "");
    authInfo.withSecret(secret);
    body.setAuthInfo(authInfo);
    request.setBody(body);
   
    AddDeviceResponse response = client.addDevice(request);
   
    // 5. 本地保存设备信息
    Device device = BeanUtil.toBean(dto, Device.class);
    device.setSecret(secret);
    device.setIotId(response.getDeviceId());
    save(device);
}

3.2 查询设备影子数据

设备影子存储了设备最新上报的数据：

@Override
public AjaxResult queryServiceProperties(String iotId) {
    ShowDeviceShadowRequest request = new ShowDeviceShadowRequest();
    request.setDeviceId(iotId);
    ShowDeviceShadowResponse response = client.showDeviceShadow(request);
   
    List<DeviceShadowData> shadow = response.getShadow();
    DeviceShadowProperties reported = shadow.get(0).getReported();
    JSONObject jsonObject = JSONUtil.parseObj(reported.getProperties());
   
    // 组装数据并返回
    List<Map<String, Object>> list = new ArrayList<>();
    jsonObject.forEach((k, v) -> {
        Map<String, Object> map = new HashMap<>();
        map.put("functionId", k);
        map.put("value", v);
        map.put("eventTime", eventTime);
        list.add(map);
    });
    return AjaxResult.success(list);
}

四、AMQP 异步消息接收

4.1 为什么用 AMQP？

优势	说明
实时性	数据秒级到达
可靠性	消息持久化，不会丢失
解耦	设备和业务系统完全解耦

4.2 AMQP 客户端实现

@Slf4j
@Component
public class AmqpClient {
    // 线程池：异步处理消息
    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
   
    @PostConstruct
    public void init() throws Exception {
        String connectionUri = buildConnectionUri();
        JmsConnectionFactory factory = new JmsConnectionFactory(connectionUri);
        connection = factory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        MessageConsumer consumer = session.createConsumer(new JmsQueue(queueName));
        consumer.setMessageListener(messageListener);
    }
   
    // 消息监听器（核心）
    private final MessageListener messageListener = message -> {
        executorService.submit(() -> processMessage(message));
    };
}

4.3 批量保存设备数据

@Override
public void batchInsertDeviceData(IotMsgNotifyData iotMsgNotifyData) {
    String iotId = iotMsgNotifyData.getHeader().getDeviceId();
    Device device = deviceMapper.selectOne(
        Wrappers.<Device>lambdaQuery().eq(Device::getIotId, iotId));
   
    iotMsgNotifyData.getBody().getServices().forEach(service -> {
        Map<String, Object> properties = service.getProperties();
        List<DeviceData> list = new ArrayList<>();
        properties.forEach((k, v) -> {
            DeviceData deviceData = BeanUtil.toBean(device, DeviceData.class);
            deviceData.setId(null);
            deviceData.setAlarmTime(eventTime);
            deviceData.setFunctionId(k);
            deviceData.setDataValue(v + "");
            list.add(deviceData);
        });
        saveBatch(list);
    });
}

五、数据库连接池优化

当设备数量增多时，可能出现连接池耗尽的问题：

错误提示：wait millis 60003, active 20, maxActive 20, creating 0

解决方案：

spring:
  datasource:
    druid:
      initialSize: 5
      minIdle: 10
      maxActive: 60           # 提升最大连接数
      maxWait: 120000         # 延长等待时间至 2 分钟
      connectTimeout: 30000
      socketTimeout: 60000
      testOnBorrow: true      # 借出连接时校验有效性

六、设备数据展示

6.1 分页查询设备数据

@Override
public TableDataInfo selectDeviceDataList(DeviceDataPageReqDto dto) {
    LambdaQueryWrapper<DeviceData> wrapper = new LambdaQueryWrapper<>();
    Page<DeviceData> page = new Page<>(dto.getPageNum(), dto.getPageSize());
   
    if (StringUtils.isNotEmpty(dto.getDeviceName())) {
        wrapper.eq(DeviceData::getDeviceName, dto.getDeviceName());
    }
    if (StringUtils.isNotEmpty(dto.getFunctionId())) {
        wrapper.eq(DeviceData::getFunctionId, dto.getFunctionId());
    }
    if (dto.getStartTime() != null && dto.getEndTime() != null) {
        wrapper.between(DeviceData::getAlarmTime, dto.getStartTime(), dto.getEndTime());
    }
   
    page = page(page, wrapper);
    return getTableDataInfo(page);
}