TDengine+MQTT：实现物联网设备数据采集

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class MqttPublisher {
    // MQTT代理服务器地址
    private static final String SERVER_URI = "tcp://mqtt-broker-address:1883"; 
    // MQTT客户端ID
    private static final String CLIENT_ID = "mqtt-client-id"; 
   
    public void publishData(String topic, String data) {
        try {
            // 创建MQTT客户端
            MqttClient mqttClient = new MqttClient(SERVER_URI, CLIENT_ID, new MemoryPersistence());
            
            // 配置连接选项
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            
            // 连接MQTT代理服务器
            mqttClient.connect(options);
            
            // 创建MQTT消息
            MqttMessage message = new MqttMessage(data.getBytes());
            message.setQos(1); // 设置QoS级别为1
            
            // 发布消息
            mqttClient.publish(topic, message);
            
            // 断开连接
            mqttClient.disconnect();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
复制

import com.taosdata.jdbc.TSDBDriver;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
public class TDengineStorage {
    private static final String DB_URL = "jdbc:TAOS://tdengine-server-address:6030/iot_db"; // TDengine数据库URL
    private static final String USER = "root"; // TDengine数据库用户名
    private static final String PASSWORD = "taosdata"; // TDengine数据库密码
    public void storeData(String deviceName, String data) {
        try {
            // 加载TDengine JDBC驱动
            Class.forName(TSDBDriver.class.getName());
            
            // 连接TDengine数据库
            Connection conn = DriverManager.getConnection(DB_URL, USER, PASSWORD);
            
            // 创建SQL语句
            String sql = "INSERT INTO " + deviceName + " (ts, data) VALUES (?, ?)";
            PreparedStatement pstmt = conn.prepareStatement(sql);
            
            // 设置参数
            pstmt.setTimestamp(1, new java.sql.Timestamp(System.currentTimeMillis()));
            pstmt.setString(2, data);
            
            // 执行SQL语句
            pstmt.executeUpdate();
            
            // 关闭连接
            pstmt.close();
            conn.close();
        } catch (ClassNotFoundException | SQLException e) {
            e.printStackTrace();
        }
    }
}
复制