修改MQTT消息策略

share-quartz/src/main/java/share/quartz/task/RedisTask.java

@@ -417,6 +417,9 @@ public class RedisTask {
                                 dvo.setRefreshNum(dvo.getRefreshNum() + 1);
                                 // 将失败的消息放到redis最后重新发送
                                 redisUtil.lPush(key2, dvo);
+                                //次数超过3次，删除
+                            } else if (dvo != null && dvo.getRefreshNum() >= 3) {
+                                redisUtil.getRightPop(key2, 10L);
                             }
                             // 获取当前需要发送的消息
                             Object data = redisUtil.getIndex(key2, -1);

share-system/src/main/java/share/system/mqtt/MqttConfig.java

@@ -15,7 +15,6 @@ import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
 import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
 import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
-import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent;
 import org.springframework.integration.mqtt.event.MqttMessageSentEvent;
 import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
 import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
@@ -23,7 +22,6 @@ import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 
-
 import javax.annotation.Resource;
 import java.util.Date;
 
@@ -154,7 +152,7 @@ public class MqttConfig {
         //接收字节数组类型有效荷载
         defaultPahoMessageConverter.setPayloadAsBytes(true);
         adapter.setConverter(defaultPahoMessageConverter);
-        adapter.setQos(2);
+        adapter.setQos(0);
         adapter.setOutputChannel(mqttInputChannel());
         return adapter;
     }