| | |
| | | |
| | | @Override |
| | | public void run(ApplicationArguments args) { |
| | | log.info("初始化并启动mqtt......"); |
| | | log.debug("初始化并启动mqtt......"); |
| | | if (PropertiesUtil.MQTT_ENABLED) { |
| | | this.connect(); |
| | | } |
| | |
| | | if (null == client) { |
| | | client = new MqttClient(PropertiesUtil.MQTT_HOST, PropertiesUtil.MQTT_CLIENT_ID, new MemoryPersistence()); |
| | | } |
| | | log.info("--创建mqtt客户端"); |
| | | log.debug("--创建mqtt客户端"); |
| | | } catch (Exception e) { |
| | | log.error("创建mqtt客户端异常:" + e); |
| | | } |
| | |
| | | options.setKeepAliveInterval(PropertiesUtil.MQTT_KEEP_ALIVE); |
| | | // 是否清除session |
| | | options.setCleanSession(true); |
| | | log.info("--生成mqtt配置对象"); |
| | | log.debug("--生成mqtt配置对象"); |
| | | return options; |
| | | } |
| | | |
| | |
| | | */ |
| | | qos[i] = 1; |
| | | } |
| | | log.info("--设置消息发布质量"); |
| | | log.debug("--设置消息发布质量"); |
| | | return qos; |
| | | } |
| | | |
| | |
| | | public void create(MqttConnectOptions options, String[] topic, int[] qos) { |
| | | try { |
| | | client.setCallback(new MqttConsumerCallback(client, options, topic, qos)); |
| | | log.info("--添加回调处理类"); |
| | | log.debug("--添加回调处理类"); |
| | | client.connect(options); |
| | | } catch (Exception e) { |
| | | log.info("装载实例或订阅主题异常:" + e); |
| | |
| | | */ |
| | | public void subscribe(String topic, int qos) { |
| | | try { |
| | | log.info("topic:" + topic); |
| | | log.debug("topic:" + topic); |
| | | client.subscribe(topic, qos); |
| | | } catch (MqttException e) { |
| | | e.printStackTrace(); |
| | |
| | | */ |
| | | public static void publish(int qos, boolean retained, String topic, String pushMessage) { |
| | | if (client != null) { |
| | | log.info("【主题】:" + topic + "【qos】:" + qos + "【pushMessage】:" + pushMessage); |
| | | log.debug("【主题】:" + topic + "【qos】:" + qos + "【pushMessage】:" + pushMessage); |
| | | MqttMessage message = new MqttMessage(); |
| | | message.setQos(qos); |
| | | message.setRetained(retained); |
| | |
| | | token = mTopic.publish(message); |
| | | token.waitForCompletion(); |
| | | if (token.isComplete()) { |
| | | log.info("消息发送成功"); |
| | | log.debug("消息发送成功"); |
| | | } |
| | | } catch (MqttPersistenceException e) { |
| | | log.error("mqtt持久异常:" + e.getMessage()); |