使用nodejs和java异步获取MQTT数据

Question

我正在尝试开发一个发布/订阅应用程序，其中 Java 程序是发布者，nodejs 程序是订阅者。 Java 客户端连接到 MQTT 服务器并发送一些随机数据。 NodeJS 客户端应该订阅 java 客户端注册的主题。我正在从 java 客户端获取数据到 NodeJS 控制台，但我必须在网页上异步打印该数据。

这是我写的代码。

MqttPublishSample.java

public class MqttPublishSample {

    public static void main(String[] args) {

        String topic = "MQTT-Examples";
        String content = "HelloWorld";
        int qos = 2;
        int i =0;
        String broker = "tcp://localhost:1883";
        String clientId = "JavaSample";
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            System.out.println("Connecting to broker: " + broker);
            sampleClient.connect(connOpts);
            System.out.println("Connected");
            while(i < 100) 
            {                           
                int num = (int) (Math.random() * 2); 
                String text = content + num;
                System.out.println("Publishing message: "+content+ " "+ num); 
                MqttMessage message = new MqttMessage(text.getBytes());
                message.setQos(qos);
                sampleClient.publish(topic, message);
                System.out.println("Message published");
                i++;
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
            // System.out.println("Publishing message: " + content);

            /*sampleClient.disconnect();
            System.out.println("Disconnected");
            System.exit(0);*/
        } catch (MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
        }
    }
}

app.js

var express = require('express');
var app = express();

app.get('/', function (req, res) {
  client.on('connect', function() {
        console.log("Connection Successful");
        client.subscribe('MQTT-Examples');

    });
    var content;
    client.on('message', function (topic, message) {
    content = message;
      console.log(message.toString());
    });
res.render('index', { title: content });
});

app.listen(3000, function () {
  console.log('Example app listening on port 3000!');
});

我尝试在网页上打印消息，但看不到任何消息。 我需要在网页上异步打印数据。有关如何进行的任何帮助。谢谢！

Answer 1

尝试在client.on 的回调中移动res.render('index', { title: content }); 行，例如：

var express = require('express');
var app = express();

/*
 * MQTT Client
 */
var content = ""; //buffer

client.on('connect', function() {
  console.log("Connection Successful");
  client.subscribe('MQTT-Examples');
});

client.on('message', function(topic, message) {
  content += message.toString() + "<br>";
  console.log(message.toString());

});

/*
 * Express
 */
app.get('/', function(req, res) {
  res.render('index', {
    title: content
  });
});

app.listen(3000, function() {
  console.log('Example app listening on port 3000!');
});

更新

我确实更新了代码，在新代码中，您将能够在每次重新加载网页时看到收到的所有消息。

在之前的实现中是一种长轮询，资源会等待来自代理的消息。

希望对你有帮助。