使用java实现从MQTT读取数据并保存到iotdb时序数据库

以下是Java代码实现从MQTT读取数据并保存到iotdb时序数据库的示例：

1. 引入相关库文件

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

<dependency>
    <groupId>org.apache.iotdb</groupId>
    <artifactId>iotdb-jdbc</artifactId>
    <version>0.10.0</version>
</dependency>

2. 连接MQTT服务器

String brokerUrl = "tcp://localhost:1883"; //MQTT服务器地址
String clientId = "iotdb"; //客户端ID
MemoryPersistence persistence = new MemoryPersistence();
MqttClient mqttClient = new MqttClient(brokerUrl, clientId, persistence);
mqttClient.connect();

3. 订阅MQTT主题

String topic = "my/topic"; //订阅的主题
int qos = 1; //消息质量
mqttClient.subscribe(topic, qos);

4. 接收MQTT消息并保存到iotdb时序数据库

String iotdbUrl = "jdbc:iotdb://localhost:6667/"; //iotdb数据库地址
String iotdbUsername = "root"; //iotdb数据库用户名
String iotdbPassword = "root"; //iotdb数据库密码
String iotdbStorageGroup = "root.my"; //iotdb存储组
String iotdbMeasurement = "sensor1"; //iotdb测点名称
String iotdbDataType = "DOUBLE"; //iotdb数据类型
String iotdbEncoding = "PLAIN"; //iotdb编码方式
String iotdbTimeseries = iotdbStorageGroup + "." + iotdbMeasurement; //iotdb时序列名

Class.forName("org.apache.iotdb.jdbc.IoTDBDriver");
Connection connection = DriverManager.getConnection(iotdbUrl, iotdbUsername, iotdbPassword);
Statement statement = connection.createStatement();

mqttClient.setCallback(new MqttCallback() {
    @Override
    public void connectionLost(Throwable throwable) {
        System.out.println("MQTT连接断开");
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        String message = new String(mqttMessage.getPayload());
        System.out.println("收到MQTT消息：" + message);

        try {
            String sql = String.format("insert into %s(time,%s) values(%d,%s)",
                    iotdbTimeseries, iotdbDataType, iotdbEncoding,
                    System.currentTimeMillis(), message);
            statement.execute(sql);
            System.out.println("保存到iotdb成功");
        } catch (SQLException e) {
            System.out.println("保存到iotdb失败：" + e.getMessage());
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }
});

5. 断开MQTT连接

mqttClient.disconnect();

完整代码示例：

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;

public class MqttToIotdb {

    public static void main(String[] args) throws MqttException, ClassNotFoundException, SQLException {
        String brokerUrl = "tcp://localhost:1883"; //MQTT服务器地址
        String clientId = "iotdb"; //客户端ID
        MemoryPersistence persistence = new MemoryPersistence();
        MqttClient mqttClient = new MqttClient(brokerUrl, clientId, persistence);
        mqttClient.connect();

        String topic = "my/topic"; //订阅的主题
        int qos = 1; //消息质量
        mqttClient.subscribe(topic, qos);

        String iotdbUrl = "jdbc:iotdb://localhost:6667/"; //iotdb数据库地址
        String iotdbUsername = "root"; //iotdb数据库用户名
        String iotdbPassword = "root"; //iotdb数据库密码
        String iotdbStorageGroup = "root.my"; //iotdb存储组
        String iotdbMeasurement = "sensor1"; //iotdb测点名称
        String iotdbDataType = "DOUBLE"; //iotdb数据类型
        String iotdbEncoding = "PLAIN"; //iotdb编码方式
        String iotdbTimeseries = iotdbStorageGroup + "." + iotdbMeasurement; //iotdb时序列名

        Class.forName("org.apache.iotdb.jdbc.IoTDBDriver");
        Connection connection = DriverManager.getConnection(iotdbUrl, iotdbUsername, iotdbPassword);
        Statement statement = connection.createStatement();

        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                System.out.println("MQTT连接断开");
            }

            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) {
                String message = new String(mqttMessage.getPayload());
                System.out.println("收到MQTT消息：" + message);

                try {
                    String sql = String.format("insert into %s(time,%s) values(%d,%s)",
                            iotdbTimeseries, iotdbDataType, iotdbEncoding,
                            System.currentTimeMillis(), message);
                    statement.execute(sql);
                    System.out.println("保存到iotdb成功");
                } catch (SQLException e) {
                    System.out.println("保存到iotdb失败：" + e.getMessage());
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

            }
        });

        //等待MQTT消息
        while (true) {

        }

        //断开MQTT连接
        //mqttClient.disconnect();
    }
}