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

首先，需要安装以下软件：

1. Eclipse Mosquitto：MQTT Broker
2. InfluxDB：时序数据库
3. Eclipse Paho：MQTT客户端库

具体步骤如下：

1. 安装Eclipse Mosquitto和InfluxDB

在Linux上，可以使用以下命令安装：

sudo apt-get install mosquitto influxdb

2. 安装Eclipse Paho

可以使用Maven在项目中添加以下依赖：

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

3. 从MQTT读取数据并保存到时序数据库

可以使用以下Java代码从MQTT读取数据并保存到InfluxDB：

import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;

public class MqttToInfluxDB {
    public static void main(String[] args) throws MqttException {
        // MQTT连接参数
        String brokerUrl = "tcp://localhost:1883";
        String clientId = "JavaMqttClient";
        String topic = "sensor/data";

        // InfluxDB连接参数
        String influxDbUrl = "http://localhost:8086";
        String influxDbName = "test";
        String measurementName = "sensor_data";

        // 创建MQTT客户端
        IMqttClient mqttClient = new MqttClient(brokerUrl, clientId);
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setCleanSession(true);
        mqttClient.connect(connOpts);

        // 创建InfluxDB客户端
        InfluxDB influxDB = InfluxDBFactory.connect(influxDbUrl);
        influxDB.createDatabase(influxDbName);

        // 订阅MQTT主题
        mqttClient.subscribe(topic, new IMqttMessageListener() {
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                // 解析MQTT消息
                String payload = new String(message.getPayload());
                String[] parts = payload.split(",");
                double value = Double.parseDouble(parts[0]);
                long timestamp = Long.parseLong(parts[1]);

                // 创建InfluxDB数据点
                Point point = Point.measurement(measurementName)
                        .time(timestamp, java.util.concurrent.TimeUnit.MILLISECONDS)
                        .addField("value", value)
                        .build();

                // 保存数据点到InfluxDB
                BatchPoints batchPoints = BatchPoints.database(influxDbName)
                        .point(point)
                        .build();
                influxDB.write(batchPoints);
            }
        });
    }
}

这个程序将订阅`sensor/data`主题，读取从传感器发送的数据，并将数据保存到名为`test`的InfluxDB数据库中的`sensor_data`测量中。