问基于嵌入式activeMQ的websocket mqtt
EN

Stack Overflow用户

提问于 2015-04-07 00:55:33

回答 2查看 2.2K关注 0票数 0

我在tcp和websockets上设置了一个ActiveMQ代理作为MQTT，如下所示：

BrokerService broker = new BrokerService();
broker.addConnector("mqtt://localhost:1883");
broker.addConnector("ws://0.0.0.0:1884");
broker.setPersistent(false);
broker.setBrokerName("default");
broker.start();

在我的类路径中，我有以下jars (因为我不想包含slf4j )：

activemq-broker-5.11.1.jar          activemq-stomp-5.11.1.jar
activemq-client-5.11.1.jar          activemq-web-5.11.1.jar
activemq-http-5.11.1.jar            commons-codec-1.6.jar
activemq-mqtt-5.11.1.jar            geronimo-j2ee-management_1.1_spec-1.0.1.jar
activemq-openwire-legacy-5.11.1.jar     geronimo-jms_1.1_spec-1.1.1.jar
activemq-parent-5.11.1              hawtbuf-1.11.jar
activemq-spring-5.11.1.jar          mqtt-client-1.10.jar
javax.servlet-api-3.1.0.jar
javax.servlet.jsp.jstl-1.2.1.jar
javax.servlet.jsp.jstl-api-1.2.1-2.jar
jetty-all-9.2.8.v20150217.jar
slf4j-api-1.7.5
logback-classic-1.0.13
logback-core-1.0.13

当我试图与paho javascript客户端连接时，它会挂起，而我的蚊子到客户端则能够通过tcp进行连接，没有任何麻烦。

在日志中，我找到了这一行：

DEBUG o.a.a.t.ws.jetty9. MQTTSocket - Waiting for StompSocket to be properly started...

阅读代码后，我发现MQTTSocket (与StompSocket相同，两者都尝试过)正在等待一个只在doStart()中被计算下来的CountDownLatch对象。显然，doStart()从未被调用过。为什么？我怎么叫它？我遗漏了什么？

我的html / javascript：

<!DOCTYPE html>
<html>
  <head>
  <meta http-equiv="Content-Type" content="text/html;charset=utf-8"/>
  <script src="shared/mqttws31-min.js" type="text/javascript"></script>
  <script type="text/javascript">

    //sample HTML/JS script that will publish/subscribe to topics in the Google Chrome Console
    //by Matthew Bordignon @bordignon on twitter.

    var wsbroker = "127.0.0.1";  //mqtt websocket enabled broker
    var wsport = 1884 // port for above

    var client = new Paho.MQTT.Client(wsbroker, wsport, "",
        "myclientid_" + parseInt(Math.random() * 100, 10));

    client.onConnectionLost = function (responseObject) {
      console.log("connection lost: " + responseObject.errorMessage);
    };

    client.onMessageArrived = function (message) {
      console.log(message.destinationName, ' -- ', message.payloadString);
    };

    var options = {
      timeout: 300,
      //mqttVersion: 3, // to constrain the protocol to version 3
      onSuccess: function () {
        console.log("mqtt connected");
        // Connection succeeded; subscribe to our topic, you can add multile lines of these
        client.subscribe('/World', {qos: 1});


        //use the below if you want to publish to a topic on connect
        message = new Paho.MQTT.Message("Hello");
        message.destinationName = "/World";
        client.send(message);

      },
      onFailure: function (message) {
        console.log("Connection failed: " + message.errorMessage);
      }
    };

  function init() {
      client.connect(options);
  }

    </script>
  </head>
  <h1>MQTT test</h1>
  <body onload="init();">
  </body>

</html>

编辑：我还尝试了xbeans配置，并使用了activeMQ附带的经过调整的xml文件：

System.setProperty("activemq.conf", "conf");
System.setProperty("activemq.data", "data");

BrokerService broker;
try {
    broker = BrokerFactory.createBroker("xbean:activemq.xml");
    broker.start();
} catch (Exception e1) {
    e1.printStackTrace();
}

XML文件：

activemq.xml：https://dl.dropboxusercontent.com/u/9632359/activemq.xml
jetty.xml：https://dl.dropboxusercontent.com/u/9632359/jetty.xml

servlets

websocket

jetty

activemq

mqtt

回答 2

Stack Overflow用户

发布于 2015-04-27 07:37:05

我想broker.addConnector("ws://0.0.0.0:1884");应该是broker.addConnector("ws://localhost:1884");
我已经用mqtt和ws编写并成功地运行了一个websocket-server，我怀疑您在那里丢失了websocket-server

请注意jetty-all和websocket-server版本依赖于activemq-http版本。这些是我的gradle依赖项：

  compile group: "org.apache.activemq",  version: "5.11.+", name: "activemq-all"

  compile group: "org.apache.activemq",  version: "5.11.+", name: "activemq-mqtt"

  compile group: "org.apache.activemq",  version: "5.11.+", name: "activemq-http"

  compile group: "org.eclipse.jetty.aggregate",  version: "8.+", name: "jetty-all"  

  compile group: "org.eclipse.jetty.websocket",  version: "9.+", name: "websocket-server"

可以指定单个连接器属性(例如名称)：

TransportConnector连接器=新的TransportConnector()；

connector = new TransportConnector();
connector.setUri(new URI("mqtt://localhost:1883"));
connector.setName("MQTTConn");
broker.addConnector(connector);

connector = new TransportConnector();
connector.setUri(new URI("ws://localhost:1884"));
connector.setName("WSConn");
broker.addConnector(connector);