网站导航
> 技术文档 > 在物联网系统中时序数据库和关系型数据库如何使用？_采取时序库和关系库相结合的方式

在物联网系统中时序数据库和关系型数据库如何使用？_采取时序库和关系库相结合的方式

网友: 技术文档 2025-09-03 03:13:06

在物联网系统中，时序数据库（TSDB）和关系型数据库（RDBMS）的存储顺序设计需要根据数据特性、业务需求和系统架构综合考虑。以下是典型的设计方案和逻辑顺序：

1. 常见存储顺序方案

方案一：先写时序数据库，后异步同步到关系型数据库

适用场景：高频传感器数据为主，业务数据可容忍短暂延迟。
流程：

  1. MQTT Broker 接收设备原始数据（如 devices/A/temperature）。
  2. 数据首先写入时序数据库（如InfluxDB）：
    • 存储原始时间序列数据（高吞吐、低延迟）。
  3. 异步处理层（如Kafka/Flink）消费数据，处理后写入关系型数据库：
    • 提取关键状态（如最新温度值）写入MySQL的device_status表。
    • 关联设备元数据（如设备所属用户）。

优点：

  • 确保传感器数据的写入性能最大化。
  • 避免高频写入拖累关系型数据库。

示例代码（伪代码）：

# MQTT回调处理def on_mqtt_message(topic, payload): # 1. 原始数据写入InfluxDB influx.write({ \"measurement\": \"sensor_data\", \"tags\": {\"device_id\": topic.split(\'/\')[1]}, \"fields\": {\"temperature\": payload.temp}, \"time\": payload.timestamp }) # 2. 异步推送至Kafka，后续处理 kafka.produce(\"device_updates\", key=device_id, value=payload)# Kafka消费者处理业务逻辑def kafka_consumer(): for message in kafka.consume(): # 3. 关联设备元数据并写入MySQL device = mysql.query(\"SELECT * FROM devices WHERE id = ?\", message.device_id) mysql.execute( \"UPDATE device_status SET last_temp = ?, updated_at = NOW() WHERE device_id = ?\", message.temp, message.device_id )

示例代码（以下是使用Java实现的等效代码，包含MQTT回调处理、InfluxDB写入和通过Kafka异步处理写入MySQL的逻辑）：

import org.eclipse.paho.client.mqttv3.*;import com.influxdb.client.InfluxDBClient;import com.influxdb.client.WriteApi;import com.influxdb.client.domain.WritePrecision;import com.influxdb.client.write.Point;import org.apache.kafka.clients.producer.*;import org.apache.kafka.clients.consumer.*;import java.sql.*;import java.time.Instant;import java.util.Properties;public class MqttDataProcessor { // InfluxDB 配置 private final InfluxDBClient influxDBClient; // Kafka 生产者 private final KafkaProducer<String, DeviceData> kafkaProducer; // MySQL 连接 private final Connection mysqlConnection; public MqttDataProcessor(InfluxDBClient influxDBClient, KafkaProducer<String, DeviceData> kafkaProducer, Connection mysqlConnection) { this.influxDBClient = influxDBClient; this.kafkaProducer = kafkaProducer; this.mysqlConnection = mysqlConnection; } // MQTT回调处理 public IMqttMessageListener createMqttListener() { return (topic, message) -> { try { // 解析payload DeviceData data = parsePayload(topic, message.getPayload()); // 1. 原始数据写入InfluxDB writeToInfluxDB(data); // 2. 异步推送至Kafka sendToKafka(data); } catch (Exception e) { e.printStackTrace(); } }; } private DeviceData parsePayload(String topic, byte[] payload) { // 这里应该是你的实际payload解析逻辑 String deviceId = topic.split(\"/\")[1]; // 示例: 假设payload是JSON格式 {\"temp\": 25.5, \"timestamp\": 123456789} String json = new String(payload); // 实际项目中可以使用Gson/Jackson等库 double temp = Double.parseDouble(json.split(\"\\\"temp\\\":\")[1].split(\",\")[0]); long timestamp = Long.parseLong(json.split(\"\\\"timestamp\\\":\")[1].split(\"}\")[0]); return new DeviceData(deviceId, temp, Instant.ofEpochSecond(timestamp)); } private void writeToInfluxDB(DeviceData data) { try (WriteApi writeApi = influxDBClient.getWriteApi()) { Point point = Point.measurement(\"sensor_data\") .addTag(\"device_id\", data.getDeviceId()) .addField(\"temperature\", data.getTemp()) .time(data.getTimestamp(), WritePrecision.S); writeApi.writePoint(point); } } private void sendToKafka(DeviceData data) { ProducerRecord<String, DeviceData> record = new ProducerRecord<>(\"device_updates\", data.getDeviceId(), data); kafkaProducer.send(record, (metadata, exception) -> { if (exception != null) { exception.printStackTrace(); } }); } // Kafka消费者处理业务逻辑 public void startKafkaConsumer() { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, \"localhost:9092\"); props.put(ConsumerConfig.GROUP_ID_CONFIG, \"device-data-group\"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,  \"org.apache.kafka.common.serialization.StringDeserializer\"); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,  \"com.your.package.DeviceDataDeserializer\"); // 需要自定义 try (KafkaConsumer<String, DeviceData> consumer = new KafkaConsumer<>(props)) { consumer.subscribe(List.of(\"device_updates\")); while (true) { ConsumerRecords<String, DeviceData> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, DeviceData> record : records) {  // 3. 关联设备元数据并写入MySQL  updateMySQL(record.value()); } } } } private void updateMySQL(DeviceData data) { String query = \"SELECT * FROM devices WHERE id = ?\"; String update = \"UPDATE device_status SET last_temp = ?, updated_at = NOW() WHERE device_id = ?\"; try (PreparedStatement selectStmt = mysqlConnection.prepareStatement(query); PreparedStatement updateStmt = mysqlConnection.prepareStatement(update)) { // 查询设备元数据 selectStmt.setString(1, data.getDeviceId()); ResultSet rs = selectStmt.executeQuery(); if (rs.next()) { // 更新设备状态 updateStmt.setDouble(1, data.getTemp()); updateStmt.setString(2, data.getDeviceId()); updateStmt.executeUpdate(); } } catch (SQLException e) { e.printStackTrace(); } } // 设备数据DTO public static class DeviceData { private String deviceId; private double temp; private Instant timestamp; // 构造器、getter和setter public DeviceData(String deviceId, double temp, Instant timestamp) { this.deviceId = deviceId; this.temp = temp; this.timestamp = timestamp; } // 省略getter和setter... }}// 使用示例public class Main { public static void main(String[] args) throws Exception { // 初始化InfluxDB客户端 InfluxDBClient influxDBClient = InfluxDBClientFactory.create( \"http://localhost:8086\", \"token\".toCharArray(), \"org\", \"bucket\" ); // 初始化Kafka生产者 Properties kafkaProps = new Properties(); kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, \"localhost:9092\"); kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, \"org.apache.kafka.common.serialization.StringSerializer\"); kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, \"com.your.package.DeviceDataSerializer\"); // 需要自定义 KafkaProducer<String, DeviceData> kafkaProducer = new KafkaProducer<>(kafkaProps); // 初始化MySQL连接 Connection mysqlConn = DriverManager.getConnection( \"jdbc:mysql://localhost:3306/iot_db\", \"user\", \"password\" ); // 创建处理器 MqttDataProcessor processor = new MqttDataProcessor( influxDBClient, kafkaProducer, mysqlConn ); // 启动Kafka消费者线程 new Thread(processor::startKafkaConsumer).start(); // 配置MQTT客户端 MqttClient mqttClient = new MqttClient(\"tcp://broker.example.com:1883\", \"java-client\"); mqttClient.connect(); // 订阅主题并设置回调 mqttClient.subscribe(\"devices/+/data\", 0, processor.createMqttListener()); }}

注意事项:
依赖库：需要添加以下依赖：

MQTT: org.eclipse.paho.client.mqttv3

InfluxDB: com.influxdb.influxdb-client-java

Kafka: org.apache.kafka.kafka-clients

MySQL: mysql.mysql-connector-java

序列化：需要为Kafka实现DeviceData的序列化器和反序列化器。

错误处理：实际项目中需要更完善的错误处理和重试机制。

资源管理：确保正确关闭所有连接和资源。

线程安全：如果高并发场景，需要考虑线程安全问题。

方案二：双写（时序库+关系库）

适用场景：数据一致性要求高，且写入压力可控。
流程：

  1. MQTT消息同时写入时序数据库和关系型数据库（需事务或最终一致性保证）。
  2. 关系型数据库仅存储关键状态快照（如设备最新状态），而非全部原始数据。

优点：

  • 数据实时一致，适合关键业务状态（如设备告警阈值）。

挑战：

  • 需处理写入冲突（如使用分布式事务或补偿机制）。
方案三：关系型数据库为主，定期归档到时序库

适用场景：历史数据分析需求明确，但实时查询以业务数据为主。
流程：

  1. 数据先写入MySQL的device_realtime表。
  2. 定时任务将过期数据批量迁移至InfluxDB，MySQL中仅保留近期数据。

优点：

  • 简化实时业务查询（所有数据在MySQL中）。
  • 降低MySQL存储压力。

2. 存储顺序设计原则

（1）根据数据特性分层
数据层级 存储目标 数据库选择 示例 原始时序数据 高频写入、长期存储 时序数据库 每秒温度读数 状态快照 最新状态查询 关系型数据库 设备当前温度、在线状态 业务元数据 关联查询、事务操作 关系型数据库 设备所属用户、地理位置
（2）写入路径优化
  • 高频数据路径：MQTT → 时序数据库 → （可选）异步聚合后写入关系库。
  • 低频元数据路径：业务系统直接CRUD操作关系型数据库。
（3）一致性保证
  • 最终一致性：通过消息队列（如Kafka）解耦，确保数据最终同步。
  • 强一致性：使用分布式事务（如XA协议），但性能较低。

3. 典型物联网架构示例

在这里插入图片描述

关键点：

  1. 实时性要求高的数据（如传感器读数）直连时序数据库。
  2. 需要业务关联的数据（如“设备所属用户”）通过流处理关联后写入MySQL。
  3. 历史数据分析直接从时序数据库查询。

4. 选择建议

  • 优先时序数据库：若90%以上的查询是基于时间范围的聚合（如“过去24小时温度趋势”）。
  • 优先关系型数据库：若需频繁JOIN查询（如“查询设备A的所有者手机号”）。
  • 混合使用：大多数生产环境会同时使用两者，通过写入顺序设计平衡性能与功能需求。

通过合理设计存储顺序，可以同时满足物联网场景的高性能写入和复杂业务查询需求。

网络标签：

相关问题