升级新版本

yzg-util-mq/src/main/java/com/yanzuoguang/mq/service/MqService.java

@@ -9,6 +9,8 @@ import org.springframework.amqp.core.Message;
 import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 
+import java.util.function.Consumer;
+
 /**
  * 消息队列服务
  *
@@ -62,6 +64,35 @@ public interface MqService {
     @ApiOperation(value = "消息收到确认")
     void basicAck(Message message, Channel channel);
 
+
+    /**
+     * 消息收到确认
+     *
+     * @param message 收到的消息
+     * @param channel 　收到的通道
+     */
+    @ApiOperation(value = "消息收到确认")
+    void basicHandle(Message messageBody, Channel channel, String exchangeNameAndRouteKey, String message, Consumer<String> consumerMessage);
+
+
+    /**
+     * 消息收到确认
+     *
+     * @param message 收到的消息
+     * @param channel 　收到的通道
+     */
+    @ApiOperation(value = "消息收到确认")
+    void basicHandle(Message messageBody, Channel channel, String exchangeNameAndRouteKey, String message, long dedTime, Consumer<String> consumerMessage);
+
+    /**
+     * 消息收到确认
+     *
+     * @param message 收到的消息
+     * @param channel 　收到的通道
+     */
+    @ApiOperation(value = "消息收到确认")
+    void basicHandle(Message messageBody, Channel channel, String exchangeName, String routeKey, String message, long dedTime, Consumer<String> consumerMessage);
+
     /**
      * 动态注册消费者回调队列
      *

yzg-util-mq/src/main/java/com/yanzuoguang/mq/service/impl/MqServiceImpl.java

@@ -6,7 +6,9 @@ import com.yanzuoguang.mq.service.*;
 import com.yanzuoguang.mq.vo.MessageVo;
 import com.yanzuoguang.mq.vo.QueueVo;
 import com.yanzuoguang.mq.vo.req.*;
+import com.yanzuoguang.util.exception.CodeException;
 import com.yanzuoguang.util.helper.StringHelper;
+import com.yanzuoguang.util.log.Log;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
@@ -14,6 +16,8 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;
 
+import java.util.function.Consumer;
+
 /**
  * 消息队列服务实现类
  *
@@ -101,6 +105,30 @@ public class MqServiceImpl implements MqService {
         messageSendService.basicAck(message, channel);
     }
 
+    @Override
+    public void basicHandle(Message messageBody, Channel channel, String exchangeNameAndRouteKey, String message, Consumer<String> consumerMessage) {
+        this.basicHandle(messageBody, channel, exchangeNameAndRouteKey, exchangeNameAndRouteKey, message, 60 * 1000, consumerMessage);
+    }
+
+    @Override
+    public void basicHandle(Message messageBody, Channel channel, String exchangeNameAndRouteKey, String message, long dedTime, Consumer<String> consumerMessage) {
+        this.basicHandle(messageBody, channel, exchangeNameAndRouteKey, exchangeNameAndRouteKey, message, dedTime, consumerMessage)
+    }
+
+    @Override
+    public void basicHandle(Message messageBody, Channel channel, String exchangeName, String routeKey, String message, long dedTime, Consumer<String> consumerMessage) {
+        try {
+            consumerMessage.accept(message);
+        } catch (CodeException e) {
+            Log.error(MqServiceImpl.class, e);
+        } catch (Exception e) {
+            Log.error(MqServiceImpl.class, e);
+            this.message(new MessageVo(exchangeName, routeKey, message, dedTime));
+        } finally {
+            this.basicAck(messageBody, channel);
+        }
+    }
+
     /**
      * 动态注册消费者回调队列
      *