智能矿山 IoT 数据采集完整教程（Java JDK21）

---

一、整体架构

矿山设备 (MQTT) 
   ↓
AWS IoT Core (Topic: mine/sensor)
   ↓
IoT Rule → Lambda: MineSensorEventHandler
   ├─ DynamoDB: 持久化矿山设备数据
   └─ ElastiCache Redis: 缓存最新设备状态

---

二、AWS 配置详细步骤

1️⃣ VPC 配置

1. 创建 VPC，例如：

   • VPC 名称：mine-vpc

   • CIDR：10.0.0.0/16

2. 创建子网：

   • 公网子网（Public Subnet）：10.0.1.0/24（用于 NAT 或公网访问）

   • 私有子网（Private Subnet）：10.0.2.0/24（Lambda + Redis）

3. 路由表：

   • 公网子网路由表 → 0.0.0.0/0 → IGW

   • 私有子网路由表 → NAT Gateway（如果 Lambda 需要访问公网）

   • 如果 DynamoDB / S3 通过 VPC Endpoint 访问，不需要 NAT

4. 安全组：

   • Lambda SG：

     • 出站：允许访问 Redis 端口 TCP 6379

     • 出站：允许访问 DynamoDB VPC Endpoint

   • Redis SG：

     • 入站：允许 Lambda SG TCP 6379

---

2️⃣ DynamoDB 表

1. 创建表：

   • 表名：realtime_mine_sensor

   • 主键：deviceId (String)

   • 排序键（可选）：timestamp

   • 按需模式（On-demand）

2. VPC Endpoint 配置（Gateway 类型）：

   • Service Name: com.amazonaws.ap-northeast-1.dynamodb

   • VPC：选择 mine-vpc

   • Route Tables：选择私有子网路由表

   • Private DNS：开启

---

3️⃣ ElastiCache Redis

1. 类型：Redis (Cluster Mode Disabled / Enabled)

2. VPC：mine-vpc，私有子网

3. 安全组：

   • 入站：允许 Lambda SG，TCP 6379

   • 出站：默认即可

4. 记录 Redis Endpoint

---

4️⃣ Lambda 函数

1. 创建 Lambda：如果使用SAM编译上传到AWS,可省略这步,但要编辑PVC使之与Elasticache,DynamoDB在同一私有子网下

   • 名称：MineSensorEventHandler

   • 运行时：Java 21

   • Handler：com.neo.lambda.handler.MineSensorEventHandler::handleRequest

   • VPC：选择私有子网 + Lambda 安全组 (如果是修改PVC,注意不是马上生效,请查看Lambda更新状态后显示"Successfully updated "再执行接下来的步骤

   • Timeout：30 秒以上（根据消息量可调）

2. IAM Role 权限（Lambda Execution Role）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:PutItem",
        "dynamodb:GetItem",
        "dynamodb:UpdateItem"
      ],
      "Resource": "arn:aws:dynamodb:ap-northeast-1:<ACCOUNT_ID>:table/realtime_mine_sensor"
    },
    {
      "Effect": "Allow",
      "Action": [
        "ec2:CreateNetworkInterface",
        "ec2:DescribeNetworkInterfaces",
        "ec2:DeleteNetworkInterface"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "*"
    }
  ]
}

注意：Lambda 在 VPC 内调用 Redis 或其他 VPC 资源，需要 EC2 CreateNetworkInterface 权限。

---

5️⃣ AWS IoT Core

1. 创建 IoT Thing（矿山设备）

   • Thing Name：mine-sensor-device

   • Policy：

     {
       "Version": "2012-10-17",
       "Statement": [
         {
           "Effect": "Allow",
           "Action": ["iot:Publish", "iot:Subscribe", "iot:Connect", "iot:Receive"],
           "Resource": "*"
         }
       ]
     }
     

2. 创建 IoT Core Rule

   • Rule SQL：

     SELECT * FROM 'mine/sensor'
     

   • 动作 → Lambda → MineSensorEventHandler

---

6️⃣ Redis 配置注意事项

• Redis 在私有子网，需要 Lambda 安全组允许出站访问 TCP 6379

• Lambda 在 VPC 内无法直接访问公网，确保 Redis 在同一 VPC 或通过 NAT

• Redis 集群模式开启时，使用 clustercfg.xxx.amazonaws.com 连接

---

7️⃣ 其他 Endpoint 配置

服务	类型	用途
DynamoDB	Gateway	Lambda 内访问 DynamoDB 不需要 NAT
S3 (可选)	Gateway	Lambda 访问代码或数据存储
Secrets Manager (可选)	Interface	存储 Redis 密码或配置
ElastiCache	无	Redis 在 VPC 内直接访问

---

三、Java 项目结构（专业推荐）

src/main/java/com/neo/lambda
├─config          → Redis/DynamoDB 配置类
├─handler         → Lambda Handler
├─model           → MineSensorDO 数据对象
├─repository      → DynamoDB Repository
├─redis           → Redis 封装类
├─service         → 核心业务逻辑
└─exception       → 自定义异常

src/test/java/com/neo/lambda
└─handler         → Lambda Handler 测试

---

四、Java 代码示例（智能矿山版）

1️⃣ 数据对象 (MineSensorDO.java)

package com.neo.lambda.model;

import java.time.Instant;

public class MineSensorDO {
    private String mineId;
    private String deviceId;
    private Instant timestamp;
    private double temperature;
    private double humidity;
    private double vibration;
    private boolean warning;

    // getter & setter
}

---

2️⃣ DynamoDB Repository

package com.neo.lambda.repository;

import com.neo.lambda.model.MineSensorDO;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;

import java.util.HashMap;
import java.util.Map;

public class MineSensorRepository {

    private final DynamoDbClient dynamoDbClient;

    public MineSensorRepository(DynamoDbClient dynamoDbClient) {
        this.dynamoDbClient = dynamoDbClient;
    }

    public void save(MineSensorDO sensor) {
        Map<String, AttributeValue> item = new HashMap<>();
        item.put("deviceId", AttributeValue.builder().s(sensor.getDeviceId()).build());
        item.put("mineId", AttributeValue.builder().s(sensor.getMineId()).build());
        item.put("timestamp", AttributeValue.builder().s(sensor.getTimestamp().toString()).build());
        item.put("temperature", AttributeValue.builder().n(String.valueOf(sensor.getTemperature())).build());
        item.put("humidity", AttributeValue.builder().n(String.valueOf(sensor.getHumidity())).build());
        item.put("vibration", AttributeValue.builder().n(String.valueOf(sensor.getVibration())).build());
        item.put("warning", AttributeValue.builder().bool(sensor.isWarning()).build());

        dynamoDbClient.putItem(PutItemRequest.builder()
                .tableName("realtime_mine_sensor")
                .item(item)
                .build());
    }
}

---

3️⃣ Redis 封装类

package com.neo.lambda.redis;

import io.lettuce.core.RedisClient;
import io.lettuce.core.api.sync.RedisCommands;

public class RedisManager {
    private static RedisClient redisClient;
    private static RedisCommands<String, String> commands;

    public static void init(String redisUri) {
        if (redisClient == null) {
            redisClient = RedisClient.create(redisUri);
            commands = redisClient.connect().sync();
        }
    }

    public static RedisCommands<String, String> getCommands() {
        return commands;
    }
}

---

4️⃣ Lambda Handler

package com.neo.lambda.handler;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.neo.lambda.model.MineSensorDO;
import com.neo.lambda.redis.RedisManager;
import com.neo.lambda.repository.MineSensorRepository;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;

import java.util.Map;

public class MineSensorEventHandler {

    private final ObjectMapper mapper = new ObjectMapper();
    private final MineSensorRepository repository = new MineSensorRepository(DynamoDbClient.create());

    public void handleRequest(Map<String, Object> event) {
        try {
            MineSensorDO sensor = mapper.convertValue(event, MineSensorDO.class);

            // 保存到 DynamoDB
            repository.save(sensor);

            // 保存到 Redis
            RedisManager.init("redis://<REDIS_ENDPOINT>:6379");
            RedisManager.getCommands().set("mine:latest:" + sensor.getDeviceId(), mapper.writeValueAsString(sensor));

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

---

5️⃣ 测试 MQTT 消息

• Topic: mine/sensor

• 消息示例：

{
  "mineId": "mine001",
  "deviceId": "sensor123",
  "timestamp": "2026-02-13T12:00:00Z",
  "temperature": 45.2,
  "humidity": 30.5,
  "vibration": 0.12,
  "warning": false
}

---

6️⃣ 注意事项

1. Lambda 在 VPC 内：

   • 确保 SG 出站 TCP 6379 能访问 Redis

   • 确保 IAM Role 有 ec2:CreateNetworkInterface 权限

2. DynamoDB VPC Endpoint：

   • 私有子网路由表指向 Gateway Endpoint

3. Redis 初始化失败：

   • 不影响 DynamoDB 写入

4. Maven 配置 JDK21：

<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-compiler-plugin</artifactId>
  <version>3.11.0</version>
  <configuration>
    <source>21</source>
    <target>21</target>
  </configuration>
</plugin>

---