devbox
羊村懒王
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

中间件(4): mqtt收发_mqttdeliverytoken.waitforcompletion

作者：羊村懒王 | 2024-03-04 20:49:17

赞

踩

mqttdeliverytoken.waitforcompletion

参考 : http://www.mokezhan.com/67091.html

maven依赖

  1. <dependency>
  2.             <groupId>org.eclipse.paho</groupId>
  3.             <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  4.             <version>1.2.0</version>
  5.         </dependency>

发送

  1.  
  2. import org.eclipse.paho.client.mqttv3.*;
  3. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
  4.  
  5. /**
  6.  * @Author: liyue
  7.  * @Date: 2021/10/13/14:15
  8.  * @Description:
  9.  */
  10. public class Send {
  11.  
  12.     /**
  13.      * 客户端唯一标识
  14.      */
  15.     public static final String MQTT_CLIENT_ID = "mqtt_broker_test_10010_send";
  16.     private static MqttTopic topic;
  17.     private static MqttClient client;
  18.  
  19.     public static void main(String[] args) {
  20.         String serverURI, userName, password, mqttTopic;
  21.         if (args.length > 0) {
  22.             serverURI = args[0];
  23.             userName = args[1];
  24.             password = args[2];
  25.             mqttTopic = args[3];
  26.         } else {
  27.             serverURI = "tcp://1.1.1.1:9087";
  28.             userName = "admin";
  29.             password = "password";
  30.             mqttTopic = "mqtt_broker_test_10010";
  31.         }
  32.         System.out.println("============================================================================================================================================");
  33.         System.out.println(StringUtil.join("消息发送程序开始启动,配置参数size:{}, serverURI:{}, userName:{}, password:{}, mqttTopic:{}",
  34.                 args.length,serverURI,userName,password,mqttTopic));
  35.         System.out.println("============================================================================================================================================");
  36.         MqttMessage message = new MqttMessage();
  37.         try {
  38.             client = new MqttClient(serverURI, MQTT_CLIENT_ID, new MemoryPersistence());
  39.             MqttConnectOptions options = new MqttConnectOptions();
  40.             options.setCleanSession(true);
  41.             options.setUserName(userName);
  42.             options.setPassword(password.toCharArray());
  43.             options.setConnectionTimeout(10);
  44.             options.setKeepAliveInterval(20);
  45.  
  46.             topic = client.getTopic(mqttTopic);
  47.  
  48.             message.setQos(1);
  49.             message.setRetained(false);
  50.             client.connect(options);
  51.             int i = 1;
  52.             while (true) {
  53.                 String msg = DateUtil.getNow() + " 测试消息" + i;
  54.                 message.setPayload(msg.getBytes());
  55.                 MqttDeliveryToken token = topic.publish(message);
  56.                 token.waitForCompletion();
  57.                 System.out.println("发送消息:["+msg+"]");
  58.                 Thread.sleep(4000);
  59.                 i++;
  60.             }
  61.         } catch (Exception e) {
  62.             e.printStackTrace();
  63.         }
  64.     }
  65. }

接收

  1.  
  2. import org.eclipse.paho.client.mqttv3.*;
  3. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
  4.  
  5. /**
  6.  * @Author: liyue
  7.  * @Date: 2021/10/13/14:15
  8.  * @Description:
  9.  */
  10. public class Receive {
  11.  
  12.     /**
  13.      * 客户端唯一标识
  14.      */
  15.     public static final String MQTT_CLIENT_ID = "mqtt_broker_test_10010_receive";
  16.  
  17.  
  18.     private volatile static MqttClient mqttClient;
  19.     private static MqttConnectOptions options;
  20.  
  21.  
  22.     public static void main(String[] args) {
  23.         String serverURI, userName, password, mqttTopic;
  24.         if (args.length > 0) {
  25.             serverURI = args[0];
  26.             userName = args[1];
  27.             password = args[2];
  28.             mqttTopic = args[3];
  29.         } else {
  30.             serverURI = "tcp://1.1.1.1:9087";
  31.             userName = "admin";
  32.             password = "password";
  33.             mqttTopic = "mqtt_broker_test_10010";
  34.         }
  35.         System.out.println("============================================================================================================================================");
  36.         System.out.println(StringUtil.join("消息接收程序开始启动,配置参数size:{}, serverURI:{}, userName:{}, password:{}, mqttTopic:{}",
  37.                 args.length,serverURI,userName,password,mqttTopic));
  38.         System.out.println("============================================================================================================================================");
  39.         try {
  40.             // host为主机名，clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，
  41.             // MemoryPersistence设置clientid的保存形式，默认为以内存保存
  42.  
  43.             mqttClient = new MqttClient(serverURI, MQTT_CLIENT_ID, new MemoryPersistence());
  44.             // 配置参数信息
  45.             options = new MqttConnectOptions();
  46.             // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
  47.             // 这里设置为true表示每次连接到服务器都以新的身份连接
  48.             options.setCleanSession(true);
  49.             // 设置用户名
  50.             options.setUserName(userName);
  51.             // 设置密码
  52.             options.setPassword(password.toCharArray());
  53.             // 设置超时时间 单位为秒
  54.             options.setConnectionTimeout(10);
  55.             // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
  56.             options.setKeepAliveInterval(20);
  57.             // 连接
  58.             mqttClient.connect(options);
  59.             // 订阅
  60.             mqttClient.subscribe(mqttTopic);
  61.             // 设置回调
  62.             mqttClient.setCallback(new MqttCallback() {
  63.                 @Override
  64.                 public void connectionLost(Throwable throwable) {
  65.                     System.err.println("无法连接mqtt服务器");
  66.                     throwable.printStackTrace();
  67.                 }
  68.  
  69.                 @Override
  70.                 public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
  71.                     System.out.println(DateUtil.getNow()+" 收到消息,Topic:["+s+"],Message:["+mqttMessage.toString()+"]");
  72.                 }
  73.  
  74.                 @Override
  75.                 public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
  76.                     try {
  77.                         System.out.println("消息传输完成,msg:"+iMqttDeliveryToken.getMessage());
  78.                     }catch (Exception e){
  79.                         System.err.println("获取消息失败");
  80.                         e.printStackTrace();
  81.                     }
  82.                 }
  83.             });
  84.         } catch (Exception e) {
  85.             e.printStackTrace();
  86.         }
  87.     }
  88. }

依赖

StringUtil.java

  1.  
  2. /**
  3.  * @Author: liyue
  4.  * @Date: 2021/10/13/14:48
  5.  * @Description:
  6.  */
  7. public class StringUtil {
  8.  
  9.     public static String join(String str, Object... param) {
  10.         for (Object p : param) {
  11.             str = str.replaceFirst("\\{\\}", p.toString());
  12.         }
  13.         return str;
  14.     }
  15. }

DateUtil.java

  1.  
  2. import java.text.ParseException;
  3. import java.text.SimpleDateFormat;
  4. import java.util.*;
  5.  
  6. /**
  7.  * 线程安全的日期工具类
  8.  */
  9. public class DateUtil {
  10.  
  11.     public static String PATTERN_YYYYMM = "yyyyMM";
  12.     public static String PATTERN_YYYY_MM_DD = "yyyy-MM-dd";
  13.     public static String PATTERN_YYYYMMDDHHMMSS = "yyyyMMddHHmmss";
  14.     public static String PATTERN_YYYYMMDDHHMM = "yyyyMMddHHmm";
  15.     public static String PATTERN_YYYY_MM_DD_HHMMSS = "yyyy-MM-dd HH:mm:ss";
  16.     public static String PATTERN_YYYY_MM_DD_HH = "yyyy-MM-dd HH";
  17.     public static String PATTERN_YYYYMMDDHH = "yyyyMMddHH";
  18.     public static String PATTERN_YYYY_MM_DD_HH_MM = "yyyy-MM-dd HH:mm";
  19.     public static String PATTERN_YYYYMMDD = "yyyyMMdd";
  20.     public static String PATTERN_YYYYMMDDHHMMSSSSS = "yyyyMMddHHmmssSSS";
  21.     public static String PATTERN_HTTP = "EEE, dd MMM yyyy HH:mm:ss zzz";
  22.     public static String PATTERN_YYYYMMDD_CHINESE = "yyyy年MM月dd日";
  23.     public static String PATTERN_YYYYMMDD_HHMMSS_CHINESE = "yyyy年MM月dd日 HH:mm:ss";
  24.     public static String PATTERN_YYYYMMDD_SLASH = "yyyy/MM/dd";
  25.     public static String PATTERN_YYYYMMDD_HHMMSS_SLASH = "yyyy/MM/dd  HH:mm:ss";
  26.     public static String PATTERN_YYYY_MM_DD_00 = "yyyy-MM-dd 00:00:00";
  27.  
  28.     public static String UTC_PATTERN_YYYY_MM_DD_T_HH_MM_SS_Z = "yyyy-MM-dd'T'HH:mm:ss'Z'";
  29.  
  30.  
  31.     /**
  32.      * 锁对象
  33.      */
  34.     private static final Object lockObj = new Object();
  35.  
  36.     /**
  37.      * 存放不同的日期模板格式的sdf的Map
  38.      */
  39.     private static Map<String, ThreadLocal<SimpleDateFormat>> sdfMap = new HashMap<String, ThreadLocal<SimpleDateFormat>>();
  40.  
  41.     /**
  42.      * 返回一个ThreadLocal的sdf,每个线程只会new一次sdf
  43.      *
  44.      * @param pattern
  45.      * @return
  46.      */
  47.     private static SimpleDateFormat getSdf(final String pattern) {
  48.         ThreadLocal<SimpleDateFormat> tl = sdfMap.get(pattern);
  49.  
  50.         // 此处的双重判断和同步是为了防止sdfMap这个单例被多次put重复的sdf
  51.         if (tl == null) {
  52.             synchronized (lockObj) {
  53.                 tl = sdfMap.get(pattern);
  54.                 if (tl == null) {
  55.                     // 只有Map中还没有这个pattern的sdf才会生成新的sdf并放入map
  56.                     System.out.println("put new sdf of pattern " + pattern + " to map");
  57.  
  58.                     // 这里是关键,使用ThreadLocal<SimpleDateFormat>替代原来直接new SimpleDateFormat
  59.                     tl = new ThreadLocal<SimpleDateFormat>() {
  60.  
  61.                         @Override
  62.                         protected SimpleDateFormat initialValue() {
  63.                             //System.out.println("thread: " + Thread.currentThread() + " init pattern: " + pattern);
  64.                             return new SimpleDateFormat(pattern);
  65.                         }
  66.                     };
  67.                     sdfMap.put(pattern, tl);
  68.                 }
  69.             }
  70.         }
  71.  
  72.         return tl.get();
  73.     }
  74.  
  75.     /**
  76.      * 是用ThreadLocal<SimpleDateFormat>来获取SimpleDateFormat,这样每个线程只会有一个SimpleDateFormat
  77.      * 时间格式化
  78.      *
  79.      * @param date
  80.      * @param pattern
  81.      * @return
  82.      */
  83.     public static String format(Date date, String pattern) {
  84.         return getSdf(pattern).format(date);
  85.     }
  86.  
  87.  
  88.     public static String format(Date date) {
  89.         return getSdf(PATTERN_YYYY_MM_DD_HHMMSS).format(date);
  90.     }
  91.  
  92.     public static String format(long timestamp, String pattern) {
  93.         return getSdf(pattern).format(new Date(timestamp));
  94.     }
  95.  
  96.     public static String format(long timestamp) {
  97.         return getSdf(PATTERN_YYYY_MM_DD_HHMMSS).format(new Date(timestamp));
  98.     }
  99.  
  100.     /**
  101.      * 时间反格式化
  102.      *
  103.      * @param dateStr
  104.      * @param pattern
  105.      * @return
  106.      * @throws ParseException
  107.      */
  108.     public static Date parse(String dateStr, String pattern) {
  109.         Date date = null;
  110.         try {
  111.             date = getSdf(pattern).parse(dateStr);
  112.         } catch (Exception e) {
  113.  
  114.         }
  115.         return date;
  116.     }
  117.  
  118.     /**
  119.      * 本地时间转 UTC 时间字符串
  120.      *
  121.      * @param date
  122.      * @return
  123.      */
  124.     public static String localToUtcString(Date date, String pattern) {
  125.         SimpleDateFormat sdf = getSdf(pattern);
  126.         sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
  127.         return sdf.format(date);
  128.     }
  129.  
  130.     /**
  131.      * UTC 时间反格式化
  132.      *
  133.      * @param date
  134.      * @param pattern
  135.      * @return
  136.      */
  137.     public static Date utcStringToUtcDate(String date, String pattern) {
  138.         SimpleDateFormat sdf = getSdf(pattern);
  139.         sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
  140.         Date utcDate = null;
  141.         try {
  142.             utcDate = sdf.parse(date);
  143.         } catch (Exception e) {
  144.  
  145.         }
  146.         return utcDate;
  147.     }
  148.  
  149.     /**
  150.      * UTC 时间格式化
  151.      *
  152.      * @param date
  153.      * @return
  154.      */
  155.     public static String utcDateToUtcString(Date date) {
  156.         SimpleDateFormat sdf = getSdf("yyyy-MM-dd'T'HH:mm:ss'Z'");
  157.         sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
  158.         return sdf.format(date);
  159.     }
  160.  
  161.  
  162.     /**
  163.      * UTC 时间字符串转本地时间
  164.      *
  165.      * @param date
  166.      * @param pattern
  167.      * @return
  168.      */
  169.     public static Date utcStringToLocalDate(String date, String pattern) {
  170.         SimpleDateFormat sdf = getSdf(pattern);
  171.         sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
  172.         Date localDate = null;
  173.         try {
  174.             localDate = sdf.parse(date);
  175.         } catch (Exception e) {
  176.  
  177.         }
  178.         return localDate;
  179.     }
  180.  
  181.     /**
  182.      * 获取本周一的日期
  183.      *
  184.      * @return
  185.      */
  186.     public static String getMonday() {
  187.         Calendar cal = Calendar.getInstance();
  188.         // 设置一个星期的第一天，按中国的习惯一个星期的第一天是星期一
  189.         cal.setFirstDayOfWeek(Calendar.MONDAY);
  190.         // 获得当前日期是一个星期的第几天
  191.         int dayWeek = cal.get(Calendar.DAY_OF_WEEK);
  192.         if (dayWeek == 1) {
  193.             dayWeek = 8;
  194.         }
  195.         // 根据日历的规则，给当前日期减去星期几与一个星期第一天的差值
  196.         cal.add(Calendar.DATE, cal.getFirstDayOfWeek() - dayWeek);
  197.         Date mondayDate = cal.getTime();
  198.         return format(mondayDate, PATTERN_YYYY_MM_DD);
  199.  
  200.     }
  201.  
  202.     /**
  203.      * 获取本周日的日期
  204.      *
  205.      * @return
  206.      */
  207.     public static String getSunday() {
  208.         Calendar cal = Calendar.getInstance();
  209.         // 设置一个星期的第一天，按中国的习惯一个星期的第一天是星期一
  210.         cal.setFirstDayOfWeek(Calendar.MONDAY);
  211.         // 获得当前日期是一个星期的第几天
  212.         int dayWeek = cal.get(Calendar.DAY_OF_WEEK);
  213.         if (dayWeek == 1) {
  214.             dayWeek = 8;
  215.         }
  216.         // 根据日历的规则，给当前日期减去星期几与一个星期第一天的差值
  217.         cal.add(Calendar.DATE, cal.getFirstDayOfWeek() - dayWeek);
  218.         cal.add(Calendar.DATE, 4 + cal.getFirstDayOfWeek());
  219.         Date sundayDate = cal.getTime();
  220.         return format(sundayDate, PATTERN_YYYY_MM_DD);
  221.     }
  222.  
  223.     /**
  224.      * 获取某月的最后一天
  225.      *
  226.      * @param year  年份
  227.      * @param month 月份
  228.      * @return
  229.      */
  230.     public static String getLastDayOfMonth(int year, int month) {
  231.         Calendar cal = Calendar.getInstance();
  232.         //设置年份
  233.         cal.set(Calendar.YEAR, year);
  234.         //设置月份
  235.         cal.set(Calendar.MONTH, month - 1);
  236.         //获取某月最大天数
  237.         int lastDay = cal.getActualMaximum(Calendar.DAY_OF_MONTH);
  238.         //设置日历中月份的最大天数
  239.         cal.set(Calendar.DAY_OF_MONTH, lastDay);
  240.         //格式化日期
  241.         String day = format(cal.getTime(), PATTERN_YYYY_MM_DD);
  242.         return day;
  243.     }
  244.  
  245.     /**
  246.      * 获取指定区间内随机时间
  247.      * @param beginDate
  248.      * @param endDate
  249.      * @param pattern
  250.      * @return
  251.      */
  252.     public static Date randomDate(String beginDate, String endDate, String pattern) {
  253.         try {
  254.             Date start = parse(beginDate, pattern);
  255.             Date end = parse(endDate, pattern);
  256.             if (start.getTime() >= end.getTime()) {
  257.                 return null;
  258.             }
  259.             long date = random(start.getTime(), end.getTime());
  260.             return new Date(date);
  261.         } catch (Exception e) {
  262.             e.printStackTrace();
  263.         }
  264.         return null;
  265.     }
  266.  
  267.     private static long random(long begin, long end) {
  268.         long rtn = begin + (long) (Math.random() * (end - begin));
  269.         if (rtn == begin || rtn == end) {
  270.             return random(begin, end);
  271.         }
  272.         return rtn;
  273.     }
  274.     
  275.     public static void main(String[] args) throws Exception {
  276.         String s = localToUtcString(new Date(), UTC_PATTERN_YYYY_MM_DD_T_HH_MM_SS_Z);
  277.         System.err.println(s);
  278.  
  279.         Date date = utcStringToLocalDate(s, UTC_PATTERN_YYYY_MM_DD_T_HH_MM_SS_Z);
  280.  
  281.         System.err.println(format(date, PATTERN_YYYY_MM_DD_HHMMSS));
  282.  
  283.         System.err.println("--------------------------------");
  284.         String da = localToUtcString(new Date(), UTC_PATTERN_YYYY_MM_DD_T_HH_MM_SS_Z);
  285.         System.err.println(da);
  286.         Date utcDate = utcStringToUtcDate(da, UTC_PATTERN_YYYY_MM_DD_T_HH_MM_SS_Z);
  287.         System.err.println(utcDateToUtcString(utcDate));
  288.         System.err.println(getSunday());
  289.     }
  290.  
  291.  
  292.     public static String getNow() {
  293.         return getSdf(PATTERN_YYYY_MM_DD_HHMMSS).format(new Date());
  294.     }
  295. }

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号