更新项目：将 Mqtt 代理（笔记本电脑）与 IOT 客户端 android 设备连接

Question

1.[已解决]作为该领域的新手，根据 Java.Threrefore 中的 OOP，我有基本的差距。
我试图从 calculateDangerLevel 方法中获取 dangerLevel 的值，以便将其与其他值一起存储在 mysql 数据库中。 到目前为止，我设法从有效载荷中获取值：y，x，电池，烟雾，气体，温度，紫外线。我需要计算传感器的值来评估危险程度。 问题是，我不能从危险级别中获取值，而是总是得到 null。

2.[已解决]我正在尝试从客户端获取已发布的主题，以便将其存储在字符串变量“topicCheck”中，然后将其放入行：result.updateString (2, "topicCheck"); 以便区分topic和接收到的msg，放到mysql db中。问题是：我不知道发布topic的取值方法，放到result.updateString(2,"topicCheck");

3.提前致谢！

import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.InputStream;
import java.sql.*;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
import java.util.Properties;
import java.util.UUID;

public class SubscriberImpl {

    private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(SubscriberImpl.class.getName());
    private static final double SMOKE_THRESHOLD = 0.14;
    private static final double GAS_THRESHOLD = 9.15;
    private static final double TEMP_THRESHOLD = 50;
    private static final double UV_THRESHOLD = 6;
    public  static String dangerLevel;
    public  static String topicCheck;
    public  static String timeEvent;

    //calculate dangerlevel
    private static String calculateDangerLevel(double smokeSensorReading, double gasSensorReading, double tempSensorReading, double uvSensorReading) {
        String dangerLevel = "No_Risk";
        if ( (smokeSensorReading > SMOKE_THRESHOLD && gasSensorReading > GAS_THRESHOLD) ||
                (smokeSensorReading <= SMOKE_THRESHOLD && gasSensorReading > GAS_THRESHOLD &&
                        tempSensorReading <= TEMP_THRESHOLD && uvSensorReading <= UV_THRESHOLD) ||
                (smokeSensorReading > SMOKE_THRESHOLD && gasSensorReading > GAS_THRESHOLD &&
                        tempSensorReading > TEMP_THRESHOLD && uvSensorReading > UV_THRESHOLD) )
            dangerLevel = "High_Risk";
        else if (smokeSensorReading <= SMOKE_THRESHOLD && gasSensorReading <= GAS_THRESHOLD &&
                tempSensorReading > TEMP_THRESHOLD && uvSensorReading > UV_THRESHOLD)
            dangerLevel = "Medium_Risk";

        return  dangerLevel;

    }

    //Parsing data from Payload
    private static void parseMqttPayload(String payload) {

        String[] payloadTokens = payload.split(",");
        // Parse the location
        if (Objects.equals(payloadTokens[0], "null") || Objects.equals(payloadTokens[1], "null"))
            return;

        // Parse the gps reading
        double y = Double.valueOf(payloadTokens[0]);
        double x = Double.valueOf(payloadTokens[1]);

        // Parse the battery reading
        double battery = Double.valueOf(payloadTokens[2]);

        // Parse the sensor readings
        double smokeSensorReading = -1;
        try { smokeSensorReading = Double.valueOf(payloadTokens[3]); }
        catch (NumberFormatException e) {}

        double gasSensorReading = -1;
        try { gasSensorReading = Double.valueOf(payloadTokens[4]); }
        catch (NumberFormatException e) {}

        double tempSensorReading = -1;
        try { tempSensorReading = Double.valueOf(payloadTokens[5]); }
        catch (NumberFormatException e) {}

        double uvSensorReading = -1;
        try { uvSensorReading = Double.valueOf(payloadTokens[6]); }
        catch (NumberFormatException e) {}

        LOGGER.warn("y: {}", y);
        LOGGER.warn("x: {}", x);
        LOGGER.warn("battery: {}", battery);
        LOGGER.warn("smoke: {}", smokeSensorReading);
        LOGGER.warn("gas: {}", gasSensorReading);
        LOGGER.warn("temp: {}", tempSensorReading);
        LOGGER.warn("uv: {}", uvSensorReading);
        LOGGER.warn("danger: {}",calculateDangerLevel(smokeSensorReading,gasSensorReading,tempSensorReading,uvSensorReading));

        //connect,insert,update data in sql via ResultSet

        try (Connection conn = DriverManager.getConnection(
                "jdbc:mysql:// localhost:3306/mqttdemo?useSSL=false&serverTimezone=UTC&useLegacyDatetimeCode=false",
                "xxxx","xxxx"))
        {

        // SECOND_WAY:insert via ResultSet

            Statement stmt2 = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE);
            ResultSet result = stmt2.executeQuery("SELECT * FROM sensorsdata");
            result.moveToInsertRow();
            result.updateInt("id", 0);
            result.updateString(2, "topicCheck");
            result.updateDouble("cordY", y);
            result.updateDouble("cordX", x);
            result.updateDouble("battery", battery);
            result.updateDouble("sensor1", smokeSensorReading);
            result.updateDouble("sensor2", gasSensorReading);
            result.updateDouble("sensor3", tempSensorReading);
            result.updateDouble("sensor4", uvSensorReading);
            result.updateString(10, calculateDangerLevel(smokeSensorReading,gasSensorReading,tempSensorReading,uvSensorReading));
            result.updateString(11,timeEvent);
            result.insertRow();
            result.moveToInsertRow();
            //result.beforeFirst();
            result.last();
            System.out.println("id = " + result.getInt("id"));
            result.close();
            stmt2.close();
        }
        catch (SQLException e) {
            System.out.println(e);
        }

    }

    public static void main(String[] args) {

        InputStream resourcesInputStream = PublisherImpl.class.getClassLoader().getResourceAsStream("application.properties");
        Properties properties = new Properties();
        try {
            properties.load(resourcesInputStream);
        } catch (IOException e) {
            LOGGER.warn("Cannot read property ", e);
        }

        String topicProperty = properties.getProperty("topic");
        String subscriberId = UUID.randomUUID().toString();
        MqttClient subscriber = null;
        try {
            subscriber = new MqttClient(properties.getProperty("mqttbroker_url"),subscriberId);
        } catch (MqttException e) {
            LOGGER.warn(e.getMessage() + " Code: " + e.getReasonCode());
        }

        //connection mqtt
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        options.setConnectionTimeout(10);

        while (true) {
            try {
                Objects.requireNonNull(subscriber).connect(options);

                try {

                    subscriber.subscribe(topicProperty, (topic, msg) -> {
                        byte[] payload = msg.getPayload();
                        LOGGER.debug("Message received: topic={}, payload={}", topic, new String(payload));
                        parseMqttPayload(new String(payload));

                        //Calculate time of the event
                        LocalDateTime localDate = LocalDateTime.now();
                        //System.out.println(localDate);
                        DateTimeFormatter dtf = DateTimeFormatter.ofPattern("dd/MM/yy h:mm:ss");
                        // System.out.println(dtf.format(localDate));
                        timeEvent =dtf.format(localDate);

                        //Distinguish topics


                    });
                    subscriber.getTopic("topic");

                    while (true);
                } catch (MqttException e) {
                    LOGGER.warn("Cannot subscribe on {}. Code={}.{}", topicProperty, e.getReasonCode(), e.getMessage());
                }
            } catch (MqttException e) {
                LOGGER.warn("Cannot connect to MQTT Broker");
            }

            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {}
        }
    }
}

Answer 1

[已解决]

我对我的代码进行了以下更改，并且似乎可以正常工作！

1. parseMqttPayload(new String(payload,**topic**));
2. private static void parseMqttPayload(String payload,String topic)
3. result.updateString("topic", topic);