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

2022-11-05 18:38:19

MQTT moquette 的Server发布主题

  1. package com.etrip.mqtt.future;
  2. import java.net.URISyntaxException;
  3. import org.fusesource.mqtt.client.FutureConnection;
  4. import org.fusesource.mqtt.client.MQTT;
  5. import org.fusesource.mqtt.client.QoS;
  6. import org.fusesource.mqtt.client.Topic;
  7. import org.slf4j.Logger;
  8. import org.slf4j.LoggerFactory;
  9. /**
  10. *
  11. *
  12. *
  13. * 采用Future式 发布主题
  14. *
  15. * @author longgangbai
  16. */
  17. public class MQTTFutureServer {
  18. private static final Logger LOG = LoggerFactory.getLogger(MQTTFutureServer.class);
  19. private final static String CONNECTION_STRING = "tcp://192.168.208.46:1883";
  20. private final static boolean CLEAN_START = true;
  21. private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s
  22. public  static Topic[] topics = {
  23. new Topic("china/beijing", QoS.EXACTLY_ONCE),
  24. new Topic("china/tianjin", QoS.AT_LEAST_ONCE),
  25. new Topic("china/henan", QoS.AT_MOST_ONCE)};
  26. public final  static long RECONNECTION_ATTEMPT_MAX=6;
  27. public final  static long RECONNECTION_DELAY=2000;
  28. public final static int SEND_BUFFER_SIZE=2*1024*1024;//发送最大缓冲为2M
  29. public static void main(String[] args)   {
  30. MQTT mqtt = new MQTT();
  31. try {
  32. //设置服务端的ip
  33. mqtt.setHost(CONNECTION_STRING);
  34. //连接前清空会话信息
  35. mqtt.setCleanSession(CLEAN_START);
  36. //设置重新连接的次数
  37. mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);
  38. //设置重连的间隔时间
  39. mqtt.setReconnectDelay(RECONNECTION_DELAY);
  40. //设置心跳时间
  41. mqtt.setKeepAlive(KEEP_ALIVE);
  42. //设置缓冲的大小
  43. mqtt.setSendBufferSize(SEND_BUFFER_SIZE);
  44. //创建连接
  45. final FutureConnection connection= mqtt.futureConnection();
  46. connection.connect();
  47. int count=1;
  48. while(true){
  49. count++;
  50. // 用于发布消息，目前手机段不需要向服务端发送消息
  51. //主题的内容
  52. String message="hello "+count+"chinese people !";
  53. String topic = "china/beijing";
  54. connection.publish(topic, message.getBytes(), QoS.AT_LEAST_ONCE,
  55. false);
  56. System.out.println("MQTTFutureServer.publish Message "+"Topic Title :"+topic+" context :"+message);
  57. }
  58. } catch (URISyntaxException e) {
  59. // TODO Auto-generated catch block
  60. e.printStackTrace();
  61. } catch (Exception e) {
  62. // TODO Auto-generated catch block
  63. e.printStackTrace();
  64. }
  65. }
  66. }

MQTT moquette 的Client接收主题

  1. package com.etrip.mqtt.future;
  2. import java.net.URISyntaxException;
  3. import org.fusesource.mqtt.client.Future;
  4. import org.fusesource.mqtt.client.FutureConnection;
  5. import org.fusesource.mqtt.client.MQTT;
  6. import org.fusesource.mqtt.client.Message;
  7. import org.fusesource.mqtt.client.QoS;
  8. import org.fusesource.mqtt.client.Topic;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11. /**
  12. *
  13. * MQTT moquette 的Client 段用于订阅主题，并接收主题信息
  14. *
  15. * 采用Future 式 订阅主题
  16. *
  17. * @author longgangbai
  18. */
  19. public class MQTTFutureClient {
  20. private static final Logger LOG = LoggerFactory.getLogger(MQTTFutureClient.class);
  21. private final static String CONNECTION_STRING = "tcp://192.168.208.46:1883";
  22. private final static boolean CLEAN_START = true;
  23. private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s
  24. private final static String CLIENT_ID = "publishService";
  25. public  static Topic[] topics = {
  26. new Topic("china/beijing", QoS.EXACTLY_ONCE),
  27. new Topic("china/tianjin", QoS.AT_LEAST_ONCE),
  28. new Topic("china/henan", QoS.AT_MOST_ONCE)};
  29. public final  static long RECONNECTION_ATTEMPT_MAX=6;
  30. public final  static long RECONNECTION_DELAY=2000;
  31. public final static int SEND_BUFFER_SIZE=2*1024*1024;//发送最大缓冲为2M
  32. public static void main(String[] args)   {
  33. //创建MQTT对象
  34. MQTT mqtt = new MQTT();
  35. try {
  36. //设置mqtt broker的ip和端口
  37. mqtt.setHost(CONNECTION_STRING);
  38. //连接前清空会话信息
  39. mqtt.setCleanSession(CLEAN_START);
  40. //设置重新连接的次数
  41. mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);
  42. //设置重连的间隔时间
  43. mqtt.setReconnectDelay(RECONNECTION_DELAY);
  44. //设置心跳时间
  45. mqtt.setKeepAlive(KEEP_ALIVE);
  46. //设置缓冲的大小
  47. mqtt.setSendBufferSize(SEND_BUFFER_SIZE);
  48. //获取mqtt的连接对象BlockingConnection
  49. final FutureConnection connection= mqtt.futureConnection();
  50. connection.connect();
  51. connection.subscribe(topics);
  52. while(true){
  53. Future<Message> futrueMessage=connection.receive();
  54. Message message =futrueMessage.await();
  55. System.out.println("MQTTFutureClient.Receive Message "+ "Topic Title :"+message.getTopic()+" context :"+String.valueOf(message.getPayloadBuffer()));
  56. }
  57. } catch (URISyntaxException e) {
  58. // TODO Auto-generated catch block
  59. e.printStackTrace();
  60. } catch (Exception e) {
  61. // TODO Auto-generated catch block
  62. e.printStackTrace();
  63. }finally{
  64. }
  65. }
  66. }
上一篇：Highcharts属性


下一篇：linux 最近使用的命令