java - MQTT Java 发布者和 Nodejs 订阅者

Question

在我当前的项目中，我们正在尝试实现两个软件组件之间的通信，如下所示。此处，Temperaturesensor(用 JavaSE 编写)组件定期发布用 MQTT JavaSE Publisher 编写的传感器测量结果。 CalculateTemp(用 Nodejs 编写)组件已使用 MQTT Nodejs 订阅并接收数据。我的问题是——当 CalculateAvgTemp 组件接收到的数据是垃圾(可能问题是 TemperatureSensor 正在以 byte 发送数据) [] 并且在 calculateAvgTemp 组件端，数据不是从 byte[] 转换的。-- 如何将 byte[] 数据转换为 JSON格式？？？)

TemperatureSensor 的 Publish 方法如下。 请注意，在发布到 MQTT 代理之前，数据会转换为字节。

public void publish(String topicName, Object arg, Device deviceInfo) {

    DataWrapper dw = new DataWrapper();
    dw.setObject(arg);
    dw.setDevice(deviceInfo);

    java.io.ByteArrayOutputStream bstream = new java.io.ByteArrayOutputStream();
    java.io.ObjectOutputStream st;

    try {
        st = new java.io.ObjectOutputStream(bstream);
        st.writeObject(dw);
        st.flush();
    } catch (IOException e) {
        e.printStackTrace();
    }

    byte[] bytes = bstream.toByteArray();

    try {

        pub.publish(topicName, 0, bytes);
    } catch (Throwable e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
}

数据接收端，CalculateAvgTemp，代码如下:

var mqtt = require('mqtt'); // no count for NodeRED
var client = mqtt.connect('mqtt://test.mosquitto.org:1883'); 
var NUM_SAMPLE_FOR_AVG = 5;
var numSample = 0;
var tempCelcius = 0;
var currentAvg = 0;
client.subscribe('tempMeasurement'); 
client.on('message', function(topic, payload) { 
    if (topic.toString() == "tempMeasurement") {

        // Here, we may need to convert the byte[] array.
        // But I do not know-- how can I convert byte[] array to JSON.  

        var sensorMeasurement=JSON.parse(payload);

        if (numSample <= NUM_SAMPLE_FOR_AVG) {
            numSample = numSample + 1;
            if (sensorMeasurement.unitOfMeasurement == 'F') {
                tempCelcius = ((sensorMeasurement.tempValue - 32) * (5 / 9));
            } else {
                tempCelcius = sensorMeasurement.tempValue;
            }
            currentAvg = parseFloat(currentAvg) + parseFloat(tempCelcius);
            if (numSample == NUM_SAMPLE_FOR_AVG) {
                currentAvg = currentAvg / NUM_SAMPLE_FOR_AVG;
                var avgTemp = {
                    "tempValue" : parseFloat(currentAvg),
                    "unitOfMeasurement" : sensorMeasurement.unitOfMeasurement
                };
                client.publish('roomAvgTempMeasurement', JSON  
                        .stringify(avgTemp)); 
                console.log("Publishing Data roomAvgTempMeasurement "); 
                numSample = 0;
                currentAvg = 0;
            }
        }
    } 
}); 

问题可能重复:MQTT communication between nodejs and java但是，所提出的解决方案不起作用，或者我无法清楚地理解它可以解决问题。

Answer 1

ObjectOutputStream 会将对象转换为序列化 Java 对象流，这是 Java 的内部对象表示法，其他任何东西都无法理解(很容易)。

您需要使用类似 json.org 中的库构建一个 JSON 对象(或手动构建一个字符串，如链接问题所示)，该对象与 DataWrapper 对象保持相同的状态，然后将 JSON 对象输出为字符串，该字符串可以作为字节有效负载发送MQTT 消息。