MQTT的学习研究（二）moquette-mqtt 的使用之mqtt broker的启动

在MQTT 官网 （http://mqtt.org/software）中有众多MQTT的实现方式。具体参看官网，Moquette是基于Apache Mina 的模型的一个Java MQTT broker。使用过Mina的同学发现其实broker的启动过程就是一个Mina应用的启动。

在MQTT moquette 中采用MINA作为底层消息的传递方式 
 
本类的目的启动MQTT moquette Broker 的方式，
本文的源代码来自  moquette-broker-0.1-jar-with-dependencies.jar 中的server类
如果想直接启动 moquette-broker-0.1-jar-with-dependencies.jar的jar文件方式
 可以执行一些命令实现 
        java -jar moquette-broker-0.1-jar-with-dependencies.jar
 
 
google code 下载MQTT moquette Broker 地址：
    http://code.google.com/p/moquette-mqtt/
    
GIT 下载MQTT moquette client 地址：
    https://github.com/fusesource/mqtt-client

在应用程序中使用MQTT的应用：

MQTT moquette 的broker服务启动代码如下：

Java代码  

  1. package com.etrip.mqtt;
  2. import java.io.File;
  3. import java.io.IOException;
  4. import java.net.InetAddress;
  5. import java.net.InetSocketAddress;
  6. import org.apache.mina.core.service.IoAcceptor;
  7. import org.apache.mina.core.service.IoServiceStatistics;
  8. import org.apache.mina.core.session.IdleStatus;
  9. import org.apache.mina.core.session.IoSession;
  10. import org.apache.mina.filter.codec.ProtocolCodecFilter;
  11. import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
  12. import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
  13. import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
  14. import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
  15. import org.dna.mqtt.moquette.proto.ConnAckEncoder;
  16. import org.dna.mqtt.moquette.proto.ConnectDecoder;
  17. import org.dna.mqtt.moquette.proto.DisconnectDecoder;
  18. import org.dna.mqtt.moquette.proto.DisconnectEncoder;
  19. import org.dna.mqtt.moquette.proto.MQTTLoggingFilter;
  20. import org.dna.mqtt.moquette.proto.PingReqDecoder;
  21. import org.dna.mqtt.moquette.proto.PingRespEncoder;
  22. import org.dna.mqtt.moquette.proto.PubAckDecoder;
  23. import org.dna.mqtt.moquette.proto.PubAckEncoder;
  24. import org.dna.mqtt.moquette.proto.PubCompDecoder;
  25. import org.dna.mqtt.moquette.proto.PubCompEncoder;
  26. import org.dna.mqtt.moquette.proto.PubCompMessage;
  27. import org.dna.mqtt.moquette.proto.PubRecDecoder;
  28. import org.dna.mqtt.moquette.proto.PubRecEncoder;
  29. import org.dna.mqtt.moquette.proto.PubRelDecoder;
  30. import org.dna.mqtt.moquette.proto.PubRelEncoder;
  31. import org.dna.mqtt.moquette.proto.PublishDecoder;
  32. import org.dna.mqtt.moquette.proto.PublishEncoder;
  33. import org.dna.mqtt.moquette.proto.SubAckEncoder;
  34. import org.dna.mqtt.moquette.proto.SubscribeDecoder;
  35. import org.dna.mqtt.moquette.proto.UnsubAckEncoder;
  36. import org.dna.mqtt.moquette.proto.UnsubscribeDecoder;
  37. import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
  38. import org.dna.mqtt.moquette.proto.messages.DisconnectMessage;
  39. import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
  40. import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
  41. import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
  42. import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
  43. import org.dna.mqtt.moquette.proto.messages.PublishMessage;
  44. import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
  45. import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
  46. import org.dna.mqtt.moquette.server.MQTTHandler;
  47. import org.slf4j.Logger;
  48. import org.slf4j.LoggerFactory;
  49. /**
  50. *
  51. * 在MQTT moquette 中采用MINA作为底层消息的传递方式
  52. *
  53. * 本类的目的启动MQTT moquette Broker 的方式，
  54. *本文的源代码来自  moquette-broker-0.1-jar-with-dependencies.jar 中的server类
  55. * 如果想直接启动 moquette-broker-0.1-jar-with-dependencies.jar的jar文件方式
  56. * 可以执行一些命令实现
  57. *        java -jar moquette-broker-0.1-jar-with-dependencies.jar
  58. *
  59. *
  60. * google code 下载MQTT moquette Broker 地址：
  61. *    http://code.google.com/p/moquette-mqtt/
  62. *
  63. * GIT 下载MQTT moquette client 地址：
  64. *  https://github.com/fusesource/mqtt-client
  65. *
  66. * @author longgangbai
  67. *
  68. *
  69. */
  70. public class MQTTBrokerProxyServer {
  71. private static final Logger LOG = LoggerFactory.getLogger(MQTTBrokerProxyServer.class);
  72. public static final String STORAGE_FILE_PATH = System.getProperty("user.home") + File.separator + "moquette_store.hawtdb";
  73. private IoAcceptor m_acceptor;
  74. SimpleMessaging messaging;
  75. public static void main(String[] args)
  76. throws IOException
  77. {
  78. new MQTTBrokerProxyServer().startServer();
  79. }
  80. protected void startServer() throws IOException
  81. {
  82. //编码协议类编码器
  83. DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
  84. decoder.addMessageDecoder(new ConnectDecoder());//连接编码
  85. decoder.addMessageDecoder(new PublishDecoder());//发布编码
  86. decoder.addMessageDecoder(new PubAckDecoder());//发布回执编码
  87. decoder.addMessageDecoder(new PubRelDecoder());
  88. decoder.addMessageDecoder(new PubRecDecoder());//接收编码
  89. decoder.addMessageDecoder(new PubCompDecoder());
  90. decoder.addMessageDecoder(new SubscribeDecoder());//订阅编码
  91. decoder.addMessageDecoder(new UnsubscribeDecoder());//取消订阅编码
  92. decoder.addMessageDecoder(new DisconnectDecoder());//断开连接编码
  93. decoder.addMessageDecoder(new PingReqDecoder());//心跳ping请求编码
  94. //解码协议类解码器
  95. DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
  96. encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder());//连接解码
  97. encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder());//订阅通知解码
  98. encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder());//取消订阅解码
  99. encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder());//发布回执解码
  100. encoder.addMessageEncoder(PubRecMessage.class, new PubRecEncoder());//接收解码
  101. encoder.addMessageEncoder(PubCompMessage.class, new PubCompEncoder());
  102. encoder.addMessageEncoder(PubRelMessage.class, new PubRelEncoder());
  103. encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());//发布解码
  104. encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder());//心跳ping相应解码
  105. encoder.addMessageEncoder(DisconnectMessage.class,new DisconnectEncoder());//断开连接解码
  106. this.m_acceptor = new NioSocketAcceptor();
  107. //设置日志的过滤链
  108. this.m_acceptor.getFilterChain().addLast("logger", new MQTTLoggingFilter("SERVER LOG"));
  109. //设置编码的过滤链
  110. this.m_acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder));
  111. //创建业务处理器类
  112. MQTTHandler handler = new MQTTHandler();
  113. //创建一个处理消息体的消息
  114. this.messaging = SimpleMessaging.getInstance();
  115. this.messaging.init();
  116. //设置消息体
  117. handler.setMessaging(this.messaging);
  118. //设置业务处理器类
  119. this.m_acceptor.setHandler(handler);
  120. ((NioSocketAcceptor)this.m_acceptor).setReuseAddress(true);
  121. ((NioSocketAcceptor)this.m_acceptor).getSessionConfig().setReuseAddress(true);
  122. this.m_acceptor.getSessionConfig().setReadBufferSize(2048);
  123. this.m_acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);
  124. this.m_acceptor.getStatistics().setThroughputCalculationInterval(10);
  125. this.m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis());
  126. //设置端口号
  127. this.m_acceptor.bind(new InetSocketAddress(1883));
  128. //获取绑定的本地的ip地址
  129. LOG.info("Server binded"+InetAddress.getLocalHost().getHostAddress());
  130. try {
  131. Thread.sleep(100000000000000L);
  132. } catch (InterruptedException e) {
  133. // TODO Auto-generated catch block
  134. e.printStackTrace();
  135. }
  136. //销毁broker对象的各种信息
  137. Runtime.getRuntime().addShutdownHook(new Thread()
  138. {
  139. public void run() {
  140. MQTTBrokerProxyServer.this.stopServer();
  141. }
  142. });
  143. }
  144. protected void stopServer() {
  145. LOG.info("Server stopping...");
  146. this.messaging.stop();
  147. //Mina  IO 统计类
  148. IoServiceStatistics statistics = this.m_acceptor.getStatistics();
  149. statistics.updateThroughput(System.currentTimeMillis());
  150. System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", new Object[] { Long.valueOf(statistics.getReadBytes()), Double.valueOf(statistics.getReadBytesThroughput()) }));
  151. System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", new Object[] { Long.valueOf(statistics.getReadMessages()), Double.valueOf(statistics.getReadMessagesThroughput()) }));
  152. //关闭相关的会话
  153. for (IoSession session : this.m_acceptor.getManagedSessions().values()) {
  154. if ((session.isConnected()) && (!session.isClosing())) {
  155. session.close(false);
  156. }
  157. }
  158. //销毁本地IoAcceptor对象
  159. this.m_acceptor.unbind();
  160. this.m_acceptor.dispose();
  161. LOG.info("Server stopped");
  162. }
  163. }

由 以上代码可以看出，在发布订阅，心跳检测，连接断开，连接时候都需要创建相关的协议编码器对象类中添加相关的编码器对象。

MQTTHandler类为主要broker处理发布和订阅消息的业务处理器类。

IoServiceStatistics类信息统计类。主要统计在mina应用中读写信息的统计。

上面代码主要讲解MQTT moquette的启动下面主要讲述服务段发布消息和客户端订阅接收信息的实现。

时间： 2024-07-31 13:39:25

MQTT的学习研究（二）moquette-mqtt 的使用之mqtt broker的启动的相关文章

MQTT的学习研究（八）基于HTTP DELETE MQTT 订阅消息服务端使用

HTTP DELETE 订阅主题请求协议和响应协议http://publib.boulder.ibm.com/infocenter/wmqv7/v7r0/topic/com.ibm.mq.csqzau.doc/ts21240_.htm 请求响应头各个字段的含义的讲解http://publib.boulder.ibm.com/infocenter/wmqv7/v7r0/topic/com.ibm.mq.csqzau.doc/ts21250_.htm 响应错误处理http://publib.boul

MQTT的学习研究（九）基于HTTP GET MQTT 抓取消息服务端使用

官方参看文档: HTTP GET 接收主题请求协议和响应协议http://publib.boulder.ibm.com/infocenter/wmqv7/v7r0/topic/com.ibm.mq.csqzau.doc/ts21230_.htm HTTP POST 发布主题请求协议和响应协议http://publib.boulder.ibm.com/infocenter/wmqv7/v7r0/topic/com.ibm.mq.csqzau.doc/ts21220_.htm HTTP DELETE

MQTT的学习研究（七）基于HTTP POST MQTT 发布消息服务端使用

参阅官方文档 http://publib.boulder.ibm.com/infocenter/wmqv7/v7r0/topic/com.ibm.mq.csqzau.doc/ts21220_.htm HTTP POST puts a message to a queue, or a publication to a topic. The HTTPPOST Java sample is an example an HTTP POST request of a message to a queue.

C++的开源跨平台日志库glog学习研究(二)--宏的使用

上一篇从整个工程上简单分析了glog,请看C++的开源跨平台日志库glog学习研究(一),这一篇对glog的实现代码入手,比如在其源码中以宏的使用最为广泛,接下来就先对各种宏的使用做一简单分析. 1. 日志输出宏 这里我们以一条最简单的日至输出为例说明: LOG(WARNING) << "This is a warning message"; 这里LOG是一个宏,其定义如下(logging.h line 487): #define LOG(severity) COMPACT

MQTT的学习研究（六） MQTT moquette 的 Blocking API 订阅消息客户端使用

* 使用 Java 为 MQ Telemetry Transport 创建订户 * 在此任务中,您将遵循教程来创建订户应用程序.订户将针对主题创建预订并接收该预订的发布. * 提供了一个示例订户应用程序 Subscribe.Subscribe 将创建预订主题 MQTT Examples,并等待获 * 得该预订的发布,等待时间为 30 秒.订户可以创建预订并等待获得发布.它还可以接收发送至先前 * 为同一客户机标识创建的预订的发布. * MqttConnectionOptions.cleanSes

MQTT的学习研究（四）moquette-mqtt 的使用之mqtt Blocking API客户端订阅并接收主题信息

在上面两篇关于mqtt的broker的启动和mqtt的服务端发布主题信息之后,我们客户端需要订阅相关的信息并接收相关的主题信息. Java代码   package com.etrip.mqtt; import java.net.URISyntaxException; import org.fusesource.mqtt.client.BlockingConnection; import org.fusesource.mqtt.client.MQTT; import org.fusesource.

MQTT的学习研究（十二） MQTT moquette 的 Future API 消息发布订阅的实现

MQTT moquette 的Server发布主题 Java代码   package com.etrip.mqtt.future; import java.net.URISyntaxException; import org.fusesource.mqtt.client.FutureConnection; import org.fusesource.mqtt.client.MQTT; import org.fusesource.mqtt.client.QoS; import org.fuseso

MQTT的学习研究（十四） MQTT moquette 的 Callback API 消息发布订阅的实现

在moquette-mqtt中提供了回调callback模式的发布和订阅但是在订阅之后没有发现有消息接收的方法,参看moquette-mqtt中Block,Future式的发布订阅基础是callback式订阅发布,但是本人在研究源代码测试,发现 callback方式接收没有成功.所以本文中只是callback式的发布和订阅没有消息接收的过程,尚未查到原因. 采用Callback式 发布主题 Java代码   package com.etrip.mqtt.callback; import java

MQTT的学习研究（五） MQTT moquette 的 Blocking API 发布消息服务端使用

参看官方文档: http://publib.boulder.ibm.com/infocenter/wmqv7/v7r0/index.jsp?topic=/com.ibm.mq.amqtat.doc/tt00000_.htm *  Java 为 MQ Telemetry Transport 创建异步发布程序 *在此任务中,您将遵循教程来修改第一个发布程序.通过修改, *使应用程序能够发送发布而不等待传递确认信息.传递确认 *信息由您创建的回调类来接收. * * * *4.使客户机断开连接 *  a