Merge branch 'ver1.2' of http://192.168.0.204/yzg/yzg-util into ver1.3

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

@@ -15,4 +15,10 @@ public interface QueueService {
      * @param req 保存队列服务
      */
     void create(QueueVo req);
+    /**
+     * 保存接口请求日志
+     *
+     * @param req 保存队列服务
+     */
+    void create(QueueVo req,boolean isAsync);
 }

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

@@ -4,19 +4,26 @@ import com.yanzuoguang.mq.dao.BeanDao;
 import com.yanzuoguang.mq.service.QueueService;
 import com.yanzuoguang.mq.vo.QueueVo;
 import com.yanzuoguang.util.helper.StringHelper;
-import org.springframework.beans.factory.annotation.Autowired;
+import com.yanzuoguang.util.thread.ThreadHelper;
 import org.springframework.stereotype.Component;
 
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
 /**
  * 交换器服务类
  *
  * @author 颜佐光
  */
 @Component
-public class QueueServiceImpl implements QueueService {
+public class QueueServiceImpl implements QueueService, Runnable {
+    private final BeanDao beanDao;
+    private final Queue<QueueVo> queue = new ConcurrentLinkedQueue<>();
+    private boolean isAsyncRun = false;
 
-    @Autowired
-    private BeanDao beanDao;
+    public QueueServiceImpl(BeanDao beanDao) {
+        this.beanDao = beanDao;
+    }
 
     /**
      * 保存接口请求日志
@@ -26,17 +33,37 @@ public class QueueServiceImpl implements QueueService {
      */
     @Override
     public void create(QueueVo req) {
-        initBean(req);
+        initBean(req, true);
+    }
+
+    @Override
+    public void create(QueueVo req, boolean isAsync) {
+        initBean(req, isAsync);
     }
 
     /**
      * 初始化实体
      *
      * @param vo      实体相关信息
+     * @param isAsync 是否异步
      */
-    private void initBean(QueueVo vo) {
-        vo.check();
+    private void initBean(QueueVo vo, boolean isAsync) {
+        if (!isAsync) {
+            initBeanHandle(vo);
+        } else {
+            queue.add(vo);
+            if (isAsyncRun) {
+                return;
+            }
+            synchronized (this) {
+                isAsyncRun = true;
+                new Thread(this).start();
+            }
+        }
+    }
 
+    private void initBeanHandle(QueueVo vo) {
+        vo.check();
         // 创建死信队列
         if (!StringHelper.isEmpty(vo.getDedQueueName())) {
             beanDao.createQueue(vo.getDedQueueName());
@@ -50,7 +77,6 @@ public class QueueServiceImpl implements QueueService {
             beanDao.createBinding(vo.getDedExchangeName(), vo.getDedQueueName(), vo.getDedRouteKey());
         }
 
-
         // 创建当前队列,并且绑定死信队列
         beanDao.createQueue(vo.getQueueName(), vo.getDedTime(), vo.getDedExchangeName(), vo.getDedRouteKey());
         // 创建当前交换器
@@ -58,4 +84,24 @@ public class QueueServiceImpl implements QueueService {
         // 创建绑定队列
         beanDao.createBinding(vo.getExchangeName(), vo.getQueueName(), vo.getRouteKey());
     }
+
+    @Override
+    public void run() {
+        while (!this.queue.isEmpty()) {
+            QueueVo vo = this.queue.poll();
+            try {
+                if (vo != null) {
+                    initBeanHandle(vo);
+                }
+            } catch (Exception ex) {
+                this.queue.add(vo);
+                ex.printStackTrace();
+                ThreadHelper.sleep(5000);
+            }
+        }
+        synchronized (this) {
+            this.isAsyncRun = false;
+        }
+    }
 }
+