MQTT协议之Apache Apollo 安装使用

MQTT协议之Apache Apollo 安装使用

Apache Apollo是一个代理服务器，其是在ActiveMQ基础上发展而来的，可以支持STOMP, AMQP, MQTT, Openwire, SSL, and WebSockets 等多种协议。

Apollo允许客户端通过开放的MQTT协议连接。该协议主要是用在资源有限的驱动上，以及网络不稳定的情况下使用，是一个订阅、发布模型。这种驱动通常不适用类似http，stomp这类基于文本，或者类似openfire，AMQP等传统二进制协议。MQTT是一个简介的二进制协议，适用这类驱动资源受限，而且是不稳定的网络条件下。之前的稳定发布版本中，MQTT是作为一个Apollo的一个插件提供的。但是现在，这个插件已经变为开发项目的一部分。MQTT在Apollo中已经不需要其他配置文件或者是第三方插件支持了。

MQTT是一个线路层的协议，任何实现该协议的客户端都可以连接到Apollo。当然也可以整合其他MQTT兼容的消息代理中。

更多有关MQTT协议内容，参考the MQTT Specification

MQTT协议配置

为了开始使用MQTT协议，首先使用MQTT3.1协议的客户端，连接到Apollo正在监听端口。Apollo会做协议检测，而且自动识别MQTT连接，而且将连接作为MQTT协议处理。你不必要为MQTT协议打开一个端口（STomp，Openfire，AMQP等都是自动识别）。如果你一定指定连接的协议，有下面两种方式：你可以选择不用协议识别，而是为MQTT指定连接：

<connector id=\protocol=\

或者你可以限制哪种协议可以被自动识别。通过下面的<detece>配置方式：

<connector id=\

<detect protocols=\

</connector>

<detect> 下protocols 对应的参数通过空格来隔开支

持的通信协议。如果只支持一种协议，就不要空格，默认情况下对任何协议生效。

如果你想调整MQTT默认设置，在apollo.xml文件中有一个<connector> 元素，通过MQTT参数配置：

<connector id=\

<mqtt max_message_length=\

</connector>

MQTT元素支持下面几个参数：

max_message_length : The size (in bytes) of the largest message that can be sent to the broker. Defaults to 100MB（broker能接受的最大消息量：默认是100M）

protocol_filters : A filter which can filter frames being sent/received to and from a client. It can modify the frame or even drop it.(一个控制发送和接收，Client的过滤器框架。可以修改，删除这个框架)

die_delay : How long after a connection is deemed to be “dead” before the connection actually closes; default: 5000ms（在实际断开连接之前，会有默认5000ms的时间被认为连接已经dead）

mqtt 配置元素也可以用来控制目的消息头的解析。下面是支持的参数：

queue_prefix : a tag used to identify destination types; default: null（用来确认目的地类型）

path_separator : used to separate segments in a destination name; default: /（用来分割目的地名称）

any_child_wildcard : indicate all child-level destinations that match the wildcard; default: +（识别子目录）

any_descendant_wildcard : indicate destinations that match the wildcard recursively; default:#（目标地址通配符）

regex_wildcard_start : pattern used to identify the start of a

regex（表示正则表达开始）

regex_wildcard_end : pattern used to identify the end of a regex（表示正则表达结束）

part_pattern : allows you to specify a regex that constrains the naming of topics. （你可以指定正则表达规则）default: [ a-zA-Z0-9\\_\\-\\%\\~\\:]+

Client 可用函数库

Apollo 支持MQTT3.1 协议，下面是可用的Clients:

Java : mqtt-client, MeQanTT

C : libmosquitto

Erlang : erlmqtt, my-mqtt4erl

.NET : MQTTDotNet, nMQTT

Perl : net-mqtt-perl,

[anyevent-mqtt-perl]https://github.com/beanz/anyevent-mqtt-perl()

Python : nyamuk

Ruby : mqtt-ruby, ruby-em-mqtt

JavaScript : Node.js MQTT Client

Delphi : TMQTTCLient

Device specific: Arduino, mbed, Nanode, Netduino

如果要找到新支持的Clients ，可以检索：the MQTT website for its software

在目录example 目录下，你可以找到一些例子，实现了与broker之间收发。

connecting

为了确保broker配置文件的安全，所以只允许一个admin 用

户连接，默认的用户名和密码是：admin ,password.

Mqtt 客户端不能specify 虚拟主机（更多请看：see the section on Virtual Hosts in the user guide)，以至于默认情况下虚拟主机已经被使用了。通常第一虚拟主机定义在apollo.xml文件中。

Destination 类型

MQTT协议是订阅，发布协议，是不允许真正的利用队列点对点的消息收发。因此Apollo仅允许利用主题，还进行MQTT消息发送。订阅的概念和持久的主题订阅 和其他协议提到的有些类似，同时也被MQTT CONNECT 框架的clean session属性控制。

Clean Sessions

但一个Client 发送一个连接，这个连接中clean session 被设置为false,那么之前连接中有相同Client_id 的session 将会被重复使用。这就意味着Client断开了，订阅依然能收到消息。这就等同与同Apollo建立一个长订阅。

如果 clean session 设置为true ，那么新session就开始了，其他的session会慢慢消失，删除。这就是Apollo中定义的普通的主题订阅。

Topic Retained Messages

如果消息被发布的同时retain 标记被设置，消息将被主题记住，以至于新的订阅到达，最近的retain 消息会被发送到订阅者。比如说：你想发布一个参数，而且你想让最新的这个参数发布到总是可用的订阅了这个主题的客户端上，你就设置在PUBLISH 框架上设置retain 标签。

注意：retained 消息 不会被设置成retained 在 QoS设置为零的broker 重启过程中。

Last Will and Testament Message

当Client第一次连接的时候，有一个will 消息和一个更QoS相关的消息会跟你有关。will消息是一个基础消息，这个基础消息只有在连接异常或者是掉线的时候才会被发送。一般用在你有一个设备，当他们掉了的时候，你需要知道。所以如果一个医疗Client从broker掉线，will消息将会作为一个

闹钟主题发送，而且会被系统作为高优先级提醒。

Reliable Messaging

MQTT协议允许Client 发布消息的时候指定Qos参数：

At Most Once (QoS=0)

At Least Once (QoS=1)

Exactly Once (QoS=2)

最多一次

这个设置时推送消息给Client，可靠性最低的一种。如果设置Qos=0，那broker就不会返回结果码，告诉你他收到消息了，也不会在失败后尝试重发。这有点像不可靠消息，如JMS。

至少一次

该设置会确保消息会被至少一次推送到Client。如果推送设

置为至少推送一次，Apollo会返回一个回调函数，确保代理已经收到消息，而且确保会确保推送该消息。如果Client 将发布了一个Qos=1的消息，如果在指定的时间内没有收到回复，Client会希望重新发布这个消息。所以可能存在这种情况：代理收到一个需要推送的消息，然后又收到一个消息推送到同一个Client。所以如果传输过程中PUBACK丢失，Client会重新发送，而且不会去检测是否是重发，broker就将消息发送到订阅主题中。

恰好一次

该设置是可靠等级最高的。他会确保发布者不仅仅会推送，而且不会像Qos=1 那样，会被接收两次。当然这个设置会增加网络的负载。当一个消息被发布出去的时候，broker会保存该消息的id,而且会利用任何长连接，坚持要把该消息推送给目标地址。如果Client收到PUBREC 标志，那就表明broker已经收到消息了。 这个时候broker会期待Client发送一个PUBREL 来清除session 中消息id，broker如果发送成功就会发送一个PUBCOMP通知Client。

Wildcard Subscriptions

通配用在主题的目标地址中。这能实现一个主题发送到多个用户，或者多层用户中。

/ is used to separate names in a path（分割路径）

+ is used to match any name in a path（通配地址任何字符）

# is used to recursively match path names（递归通配）

比如通配可能这样来用：

PRICE/# : Any price for any product on any exchange（任何交易中任何产品的价格）

PRICE/STOCK/# : Any price for a stock on any exchange（任何交易中的股票价格）

PRICE/STOCK/NASDAQ/+ : Any stock price on NASDAQ（纳斯达克的任何股票价格）

PRICE/STOCK/+/IBM : Any IBM stock price on any exchange（任何交易中IBM股票价格）

Keep Alive

Apollo只有在Client指定了CONNECT的KeepAlive 值的时候，才会设置保持连接、心跳检测。如果one Client指定了keepalive,apollo 将会使用1.5*keepalive值。这个在MQTT中有说明。

Destination Name Restrictions

路径名称限制了使用（a-z, A-Z, 0-9, _, - %, ~, :, ' ', '(', ')' ，. ）字符，通配符（*）在复杂的分隔符中。而且确保使用utf-8来编译你的URL。源码地址：

http://activemq.apache.org/apollo/community/source.html

官网参考手册：

http://activemq.apache.org/apollo/versions/1.7/website/index.html

svn checkOut :

http://svn.apache.org/repos/asf/activemq/activemq-apollo/trunk1、下载，安装(安装需要java环境)

wget

http://www.apache.org/dyn/closer.cgi?path=activemq/activemq-apollo/1.7/apache-apollo-1.7-unix-distro.tar.gz tar -zxvf apache-apollo-1.7-unix-distro.tar.gz cd apache-apollo-1.7 新建一个broker实例

./bin/apollo create mqtt_test #在当前目录下生成一个mqtt_test目录,其下面包含: bin 运行脚本 etc 环境配置 data 存储持久化数据 log 运行日志 tmp 临时文件

注：etc下配置文件说明： 一、users.properties：

用来配置可以使用服务器的用户以及相应的密码。

其在文件中的存储方式是：用户名=密码，如：

admin=password

表示新增一个用户，用户名是：admin，密码是：password

二、groups.properties：

持有群体的用户映射，可以通过组而不是单个用户简化访问控制列表。

可以为一个定义的组设置多个用户，用户之间用“|”隔开，如：

admins=admin|lily

表示admins组中有admin和lily两个用户

三、black-list.txt：

用来存放不允许连接服务器的IP地址，相当于黑名单类似的东西。

例如：10.20.9.147

表示上面IP不能够连接到服务器。

四、login.config：

是一个服务器认证的配置文件，为了安全apollo1.6版本提供了认证功能，只有相应的用户名和正确的密码才能够连接

服务器。

五、服务器主配置文件apollo.xml：

该配置文件用于控制打开的端口，队列，安全，虚拟主机设置等。

1、认证：可以使用

<authenticationdomain=\来配置是否需要连接认证，如果将其属性enable设置为false表示不用认证，任何人都可以连接服务器，默认为true

2、access_rule：可以在broker或者virtual_host中用于定义用户对服务器资源的各种行为。如：

<access_rule allow=\destroy send receive consume\表示群组users里面的

用户可以对服务器资源进行的操作有：connect 、create、 destroy、 send 、receive 、consume。详细的操作说明见：

http://activemq.apache.org/apollo/documentation/user-manual.html

3、message stores：默认情况下apollo使用的是LevelDB store，当使用关闭命令关闭不了apollo时可以使用kill进行杀死进程，但此命令有可能会导致data目录的数据文件损坏。表现为队列里有消息，但是消费不了，因此建议换成bdb存储。使用LevelDB store的配置是：

<leveldb_store directory=\默认有提供不用任何修改。使用BDB store需要到网站下jar包支持

http://download.oracle.com/maven/com/sleepycat/je/5.0.34/je-5.0.34.jar，将jar包放在服务器的lib目录下面，然后将配置文件改成：<bdb_store

directory=\即可。注意：使用bdb后可以修复kill进程导致数据损坏问题，但是bdb修复需要时间，使用者启动后需要等待修复才能消费消息 关于bdb使用可参考apollo文档

4、connector：用于配置服务器支持的链接协议以及

相应的端口。如：

<connector id=\connection_limit=\表示支持tcp链接，使用的端口是61613，链接限制是2000，自动侦听的协议是mqtt协议。

具体查看：

http://activemq.apache.org/apollo/documentation/user-manual.html 2、配置、启动

配置参考如上说明。若非本机安装，修改控制台web_admin的ip地址，以便访问控制台。

<web_admin bind=\

<web_admin bind=\启动服务: ./bin/apollo-broker-service start脚本参数：

apollo-broker-service {start|stop|restart|force-stop|status} 启动后，访问http://192.168.36.102:61680/或者https://192.168.36.102:61681。默认用户:admin / password3、测试

(1)使用paho的mqttv3测试

服务端：

[java] view plain copy print?package cn.smartslim.mqtt.demo.paho; import

org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.MqttTopic; import org.eclipse.paho.client.mqttv3.internal.MemoryPersistence; public class ApolloServer { private static String host = \ private static String userName = \ private static String passWord = \ private static MqttClient client; private static MqttTopic topic; private static MqttMessage message; private static String topicStr = \ public static void main(String[] args) throws MqttException { // host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示， //

MemoryPersistence设置clientid的保存形式，默认为以内

存保存 client = new MqttClient(host,

\

MqttConnectOptions options = new MqttConnectOptions(); // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录， // 这里设置为true表示每次连接到服务器都以新的身份连接 options.setCleanSession(true); // 设置连接的用户名 options.setUserName(userName); // 设置连接的密码

options.setPassword(passWord.toCharArray()); // 设置超时时间 单位为秒

options.setConnectionTimeout(10); // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制 options.setKeepAliveInterval(20); client.setCallback(new MqttCallback()

{ public void messageArrived(MqttTopic topicName, MqttMessage message) throws Exception { //subscribe后得到的消息会执行到这里面

System.out.println(\ System.out.println(topicName+\

} public void deliveryComplete(MqttDeliveryToken token)

{ //publish后会执行到这里 System.out.println(\ + token.isComplete()); } public void connectionLost(Throwable cause)

{ // //连接丢失后，一般在这里面进行重连

System.out.println(\ } }); topic =

client.getTopic(topicStr); message = new

MqttMessage(); message.setQos(1); message.setRetained(true);

System.out.println(message.isRetained()+\状态\

message.setPayload(\ client.connect(options); MqttDeliveryToken token = topic.publish(message); token.waitForCompletion();

System.out.println(token.isComplete()+\ } } 客户端：

[java] view plain copy print?package cn.smartslim.mqtt.demo.paho; import

org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.MqttTopic; import org.eclipse.paho.client.mqttv3.internal.MemoryPersistence; public class ApolloClient { private static String host = \ private static String userName = \ private static String passWord = \ private static MqttClient client; private static String topicStr = \ public static void main(String[] args) throws MqttException { //host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示， //MemoryPersistence设置clientid的保存形式，默认为以内存保存 client = new MqttClient(host, \

MqttConnectOptions options = new MqttConnectOptions();

//设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录， //这里设置为true表示每次连接到服务器都以新的身份连接 options.setCleanSession(true); //设置连接的用户名 options.setUserName(userName); //设置连接的密码

options.setPassword(passWord.toCharArray()); // 设置超时时间 单位为秒

options.setConnectionTimeout(10); // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制 options.setKeepAliveInterval(20); //回调 client.setCallback(new MqttCallback() { public void messageArrived(MqttTopic topicName, MqttMessage message) throws Exception { //subscribe后得到的消息会执行到这里面

System.out.println(\ System.out.println(topicName+\ } public void deliveryComplete(MqttDeliveryToken token)

{ //publish后会执行到这里

System.out.println(\ + token.isComplete()); } public void connectionLost(Throwable cause)

{ // //连接丢失后，一般在这里面进行重连

System.out.println(\ } }); //链接 client.connect(options); //订阅 client.subscribe(topicStr, 1); } } （2）使用fusesource的Callback阻塞式发布订阅消息 发布消息：

[java] view plain copy print?package

cn.smartslim.mqtt.demo.fusesource; import org.fusesource.mqtt.client.Callback; import

org.fusesource.mqtt.client.CallbackConnection; import org.fusesource.mqtt.client.MQTT; import org.fusesource.mqtt.client.QoS; import org.fusesource.mqtt.client.Topic; public class CallbackServer { private static String HOST = \ private static String USERNAME = \ private static String PASSWORD = \ private final static

boolean CLEAN_START = true; private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s public static Topic[] topics = { new Topic(\

QoS.EXACTLY_ONCE) }; public final static long RECONNECTION_ATTEMPT_MAX = 6; public final static long RECONNECTION_DELAY = 2000; public final static int SEND_BUFFER_SIZE = 2 * 1024 * 1024;// 发送最大缓冲为2M public static void main(String[] args) throws Exception { // 创建MQTT对象 MQTT mqtt = new MQTT(); mqtt.setClientId(\ // 设置mqtt broker的ip和端口 mqtt.setHost(HOST); mqtt.setUserName(USERNAME);

mqtt.setPassword(PASSWORD); // 连接前清空会话信息

mqtt.setCleanSession(CLEAN_START); // 设置重新连接的次数

mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX); // 设置重连的间隔时间 mqtt.setReconnectDelay(RECONNECTION_DELAY); // 设置心跳时间

mqtt.setKeepAlive(KEEP_ALIVE); // 设置缓冲的大小

mqtt.setSendBufferSize(SEND_BUFFER_SIZE); // 获取mqtt的连接对象CallbackConnection CallbackConnection connection = mqtt.callbackConnection();

connection.connect(new Callback<Void>() { public void onSuccess(Void value) { System.out.println(\连接成功:\ } public void onFailure(Throwable value)

{ System.out.println(\连接失败

\ } }); //发布消息 connection.publish(\测试mqtt数据\ true, new Callback<Void>() { public void onSuccess(Void value) { //与服务器断开连接成功 System.out.println(\发送成功:\ } public void onFailure(Throwable value) { //与服务器断开连接失败 System.out.println(\发送失败\ } });

Thread.sleep(600000); } } 订阅消息： [java] view plain copy print?package

cn.smartslim.mqtt.demo.fusesource; import org.fusesource.hawtbuf.Buffer; import org.fusesource.hawtbuf.UTF8Buffer; import org.fusesource.mqtt.client.Callback; import

org.fusesource.mqtt.client.CallbackConnection; import org.fusesource.mqtt.client.Listener; import org.fusesource.mqtt.client.MQTT; import org.fusesource.mqtt.client.QoS; import org.fusesource.mqtt.client.Topic; public class CallbackClient { private static String HOST = \ private static String USERNAME = \ private static String PASSWORD = \ private final static boolean CLEAN_START = true; private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s public static Topic[] topics = { new Topic(\

QoS.EXACTLY_ONCE) }; public final static long RECONNECTION_ATTEMPT_MAX = 6; public final static long RECONNECTION_DELAY = 2000;

public final static int SEND_BUFFER_SIZE = 2 * 1024 * 1024;// 发送最大缓冲为2M public static void main(String[] args) throws Exception { // 创建MQTT对象 MQTT mqtt = new MQTT(); mqtt.setClientId(\ // 设置mqtt broker的ip和端口 mqtt.setHost(HOST); mqtt.setUserName(USERNAME);

mqtt.setPassword(PASSWORD); // 连接前清空会话信息

mqtt.setCleanSession(CLEAN_START); // 设置重新连接的次数

mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX); // 设置重连的间隔时间 mqtt.setReconnectDelay(RECONNECTION_DELAY); // 设置心跳时间

mqtt.setKeepAlive(KEEP_ALIVE); // 设置缓冲的大小

mqtt.setSendBufferSize(SEND_BUFFER_SIZE); // 获取mqtt的连接对象CallbackConnection CallbackConnection connection =

mqtt.callbackConnection(); //连接 connection.connect(new Callback<Void>()

{ //可以在connect的onSuccess方法中发布或者订阅相应的主题 public void

onSuccess(Void oid) { //进入该方法表示连接成功连接成功

System.out.println(\连接成功：\ } //onFailure方法中作相应的断开连接等操作 public void onFailure(Throwable cause)

{ //进入该方法表示连接失败 System.out.println(\连接失败

\ } }); //监听 connection.listener(new Listener() { //表示成功，可以获取到订阅的主题和订阅的内容（UTF8Buffer topicmsg表示订阅的主题, //Buffer msg表示订阅的类容），一般的可以在这个方法中获取到订阅的主

题和内容然后进行相应的判断和其他业务逻辑操作； public void onPublish(UTF8Buffer topicmsg, Buffer msg, Runnable ack) { //utf-8 is used for dealing with the garbled String topic = topicmsg.utf8().toString(); String payload = msg.utf8().toString(); System.out.println(\ System.out.println(\

//表示监听成功

ack.run(); } //表示监听失败，这里可以调用相应的断开连接等方法； public void onFailure(Throwable arg0)

{ //表示监听失败 } //表示监听到连接建立，该方法只在建立连接成功时执行一次， //表示连接成功建立，如果有必要可以在该方法中进行相应的订阅操作； public void onDisconnected() { //表示监听到断开连接 } //表示监听到连接断开，该方法只在断开连接时执行一次，如有必要可以进行相应的资源回收操作。 public void onConnected() { //表示监听到连接成功 } }); //订阅 connection.subscribe(topics, new Callback<byte[]>() { //Topic[] topics表示定于的主题数组，注意

只有在改方法订阅的主题，才能够在监听方法中接收到。 public void onSuccess(byte[] qoses)

{ //主题订阅成功 System.out.println(\订阅成功:\

String(qoses)); } public void onFailure(Throwable arg0) { //状态主

题订阅失败 System.out.println(\订阅失败\ } }); //断开连接 /*connection.disconnect(new Callback<Void>() { public void onSuccess(Void value) { //与服务器断开连接成

功 } public void onFailure(Throwable value) { //与服务器断开连接失败 } });*/ //回调将执行与连接相关联的调度队列，以便可以安全使用从回调的连接，但你绝不能在回调中执行任何阻塞操作，否则会改变执行的顺序， //这样可能出错。如果可能存在阻塞时，最好是在连接的调度队列中执行另外一个线程

connection.getDispatchQueue().execute(new Runnable() { public void run() { //在这里进行相应的订阅、发布、停止连接等等操作 } }); Thread.sleep(600000); } } （3）使用桌面测试工具，可以使用的有

org.eclipse.paho.ui.app-1.0.0-win32.win32.x86_64，TT3，谷歌浏览器chrome的MQTTLens插件等。

本文来源：https://www.bwwdw.com/article/neh6.html