udpate

components/data/MongodbData/pom.xml

@@ -15,9 +15,9 @@
 
 
     <properties>
-        <mongo-java-driver.version>3.12.14</mongo-java-driver.version>
-        <mongodb-driver-sync.version>4.11.1</mongodb-driver-sync.version>
-        <spring-data-mongodb.version>4.2.3</spring-data-mongodb.version>
+<!--        <mongo-java-driver.version>3.12.14</mongo-java-driver.version>-->
+        <mongodb-driver-sync.version>5.2.1</mongodb-driver-sync.version>
+        <!--        <spring-data-mongodb.version>4.2.3</spring-data-mongodb.version>-->
     </properties>
 
     <dependencies>
@@ -31,27 +31,33 @@
 
 
         <!-- db -->
+        <!--        <dependency>-->
+        <!--            <groupId>org.springframework.data</groupId>-->
+        <!--            <artifactId>spring-data-mongodb</artifactId>-->
+        <!--            <version>${spring-data-mongodb.version}</version>-->
+        <!--            <exclusions>-->
+        <!--                <exclusion>-->
+        <!--                    <artifactId>spring-data-commons</artifactId>-->
+        <!--                    <groupId>org.springframework.data</groupId>-->
+        <!--                </exclusion>-->
+        <!--            </exclusions>-->
+        <!--        </dependency>-->
+        <!--        <dependency>-->
+        <!--            <groupId>org.springframework.boot</groupId>-->
+        <!--            <artifactId>spring-boot-starter-data-mongodb</artifactId>-->
+        <!--            <version>${spring-boot.version}</version>-->
+        <!--            <exclusions>-->
+        <!--                <exclusion>-->
+        <!--                    <groupId>org.mongodb</groupId>-->
+        <!--                    <artifactId>*</artifactId>-->
+        <!--                </exclusion>-->
+        <!--            </exclusions>-->
+        <!--        </dependency>-->
+
+
         <dependency>
             <groupId>org.springframework.data</groupId>
             <artifactId>spring-data-mongodb</artifactId>
-            <version>${spring-data-mongodb.version}</version>
-            <exclusions>
-                <exclusion>
-                    <artifactId>spring-data-commons</artifactId>
-                    <groupId>org.springframework.data</groupId>
-                </exclusion>
-            </exclusions>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-data-mongodb</artifactId>
-            <version>${spring-boot.version}</version>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.mongodb</groupId>
-                    <artifactId>*</artifactId>
-                </exclusion>
-            </exclusions>
         </dependency>
 
         <!--        驱动-->

components/data/MongodbData/src/main/java/com/github/microservice/components/data/mongo/queue/config/ExecQueueConfig.java

@@ -16,14 +16,26 @@ public class ExecQueueConfig {
     /**
      * 队列执行的延迟时间
      */
-    public Long sleepTime(int tryIndex) {
-        return tryIndex * 1000L;
+    public long sleepTime(int tryIndex) {
+        return 3 * 1000L * tryIndex;
+    }
+
+    /**
+     * work状态的执行超时时间
+     */
+    public long executeTimeOut() {
+        return 1000L * 60;
+    }
+
+    // loop_event的间隔时间
+    public long loopEventTime() {
+        return 1000L;
     }
 
     /**
      * 最大的批量执行执行数
      */
-    private int maxExecuteBatchCount() {
+    public int maxBatchExecuteCount() {
         return 10;
     }
 

components/data/MongodbData/src/main/java/com/github/microservice/components/data/mongo/queue/dao/ExecQueueDao.java

@@ -1,8 +1,9 @@
 package com.github.microservice.components.data.mongo.queue.dao;
 
 import com.github.microservice.components.data.mongo.mongo.dao.MongoDao;
+import com.github.microservice.components.data.mongo.queue.dao.extend.ExecQueueDaoExtend;
 import com.github.microservice.components.data.mongo.queue.domain.ExecQueue;
 
-public interface ExecQueueDao extends MongoDao<ExecQueue> {
+public interface ExecQueueDao extends MongoDao<ExecQueue>, ExecQueueDaoExtend {
 
 }

components/data/MongodbData/src/main/java/com/github/microservice/components/data/mongo/queue/dao/extend/ExecQueueDaoExtend.java

@@ -9,7 +9,10 @@ public interface ExecQueueDaoExtend {
     boolean updateStatus(String id, ExecQueue.ExecQueueStatus status);
 
 
-    List<ExecQueue> scan(ExecQueue.ExecQueueStatus status, int maxCount);
+    List<ExecQueue> scanWait(int maxCount);
 
 
+    //回复待执行队列
+    long restoreWaitStatus(long executeTimeOut);
+
 }

components/data/MongodbData/src/main/java/com/github/microservice/components/data/mongo/queue/dao/impl/ExecQueueDaoImpl.java

@@ -5,6 +5,7 @@ import com.github.microservice.components.data.mongo.queue.dao.extend.ExecQueueD
 import com.github.microservice.components.data.mongo.queue.domain.ExecQueue;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.BulkOperations;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
@@ -29,10 +30,38 @@ public class ExecQueueDaoImpl implements ExecQueueDaoExtend {
     }
 
     @Override
-    public List<ExecQueue> scan(ExecQueue.ExecQueueStatus status, int maxCount) {
-        Query query = new Query(Criteria.where("status").is(status));
+    public List<ExecQueue> scanWait(int maxCount) {
+        final long workTime = this.dbHelper.getTime();
+        Query query = new Query(
+                Criteria.where("status").is(ExecQueue.ExecQueueStatus.Wait)
+                        .and("nextTime").lt(this.dbHelper.getTime())
+        );
         query.with(Sort.by(Sort.Direction.ASC, "nextTime", "workTime"));
-        query.limit(maxCount);
-        return this.mongoTemplate.find(query, ExecQueue.class);
+
+        Update update = new Update();
+        update.set("status", ExecQueue.ExecQueueStatus.Work);
+        update.set("workTime", workTime);
+
+        final BulkOperations bulkOperations = this.mongoTemplate.bulkOps(BulkOperations.BulkMode.ORDERED, ExecQueue.class);
+        for (int i = 0; i < maxCount; i++) {
+            bulkOperations.updateOne(query, update);
+        }
+        bulkOperations.execute();
+
+        return this.mongoTemplate.find(Query.query(Criteria.where("workTime").is(workTime)), ExecQueue.class);
     }
+
+    @Override
+    public long restoreWaitStatus(long executeTimeOut) {
+        Query query = new Query(
+                Criteria.where("status").is(ExecQueue.ExecQueueStatus.Work)
+                        .and("workTime").lt(this.dbHelper.getTime() - executeTimeOut)
+        );
+
+        Update update = new Update();
+        update.set("status", ExecQueue.ExecQueueStatus.Wait);
+        this.dbHelper.updateTime(update);
+        return this.mongoTemplate.updateMulti(query, update, ExecQueue.class).getModifiedCount();
+    }
+
 }

components/data/MongodbData/src/main/java/com/github/microservice/components/data/mongo/queue/domain/ExecQueue.java

@@ -5,10 +5,11 @@ import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
-import org.springframework.context.ApplicationEvent;
 import org.springframework.data.mongodb.core.index.Indexed;
 import org.springframework.data.mongodb.core.mapping.Document;
 
+import java.util.Map;
+
 @Data
 @Document
 @NoArgsConstructor
@@ -27,14 +28,11 @@ public class ExecQueue extends SuperEntity {
     @Indexed
     private int currentTryCount;
 
-    // 类名
-    private Class<? extends Object> ObjectClass;
-
     // 数据
-    private Object object;
+    private Map<String,Object> data;
 
     //事件类型
-    private Class<? extends ApplicationEvent> eventClass;
+    private String eventClass;
 
     //状态
     @Indexed
@@ -55,7 +53,9 @@ public class ExecQueue extends SuperEntity {
 
         Work("进行中"),
 
-        Done("结束"),
+        Done("完成"),
+
+        Stop("停止"),
 
         ;
 

components/data/MongodbData/src/main/java/com/github/microservice/components/data/mongo/queue/event/ExecQueueEvent.java

@@ -1,12 +1,8 @@
 package com.github.microservice.components.data.mongo.queue.event;
 
-import org.springframework.context.ApplicationEvent;
-
-public class ExecQueueEvent extends ApplicationEvent {
-
-    public ExecQueueEvent(Object source) {
-        super(source);
-    }
+import java.util.Map;
 
+public interface ExecQueueEvent {
 
+    boolean execute(Map<String,Object> o);
 }

components/data/MongodbData/src/main/java/com/github/microservice/components/data/mongo/queue/service/ExecQueueService.java

@@ -4,11 +4,13 @@ import com.github.microservice.components.data.mongo.mongo.helper.DBHelper;
 import com.github.microservice.components.data.mongo.queue.config.ExecQueueConfig;
 import com.github.microservice.components.data.mongo.queue.dao.ExecQueueDao;
 import com.github.microservice.components.data.mongo.queue.domain.ExecQueue;
+import com.github.microservice.components.data.mongo.queue.event.ExecQueueEvent;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.ApplicationEvent;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
 
+import java.util.Map;
+
 @Component
 public class ExecQueueService {
 
@@ -23,27 +25,23 @@ public class ExecQueueService {
     private DBHelper dbHelper;
 
     // 添加
-    public String add(Object o, int maxTryCount, Class<? extends ApplicationEvent> eventClass) {
+    public String add(Map<String, Object> data, int maxTryCount, Class<? extends ExecQueueEvent> eventClass) {
         Assert.state(maxTryCount > 0, "最大尝试次数必须大于0");
         Assert.notNull(eventClass, "事件不能为空");
 
         final ExecQueue execQueue = new ExecQueue();
 
-        execQueue.setObject(o);
-        execQueue.setObjectClass(o.getClass());
-        execQueue.setEventClass(eventClass);
+        execQueue.setData(data);
+        execQueue.setEventClass(eventClass.getName());
 
-        execQueue.setCurrentTryCount(1);
+        execQueue.setCurrentTryCount(0);
         execQueue.setMaxTryCount(maxTryCount);
         execQueue.setStatus(ExecQueue.ExecQueueStatus.Wait);
 
-
-
         // 更新下次执行时间
         this.updateNextExecTime(execQueue);
 
         this.execQueueDao.save(execQueue);
-
         return execQueue.getId();
     }
 
@@ -58,13 +56,14 @@ public class ExecQueueService {
      * 更新下次执行时间
      */
     public boolean updateNextExecTime(ExecQueue execQueue) {
-        if (execQueue.getCurrentTryCount() > execQueue.getMaxTryCount()) {
+        if (execQueue.getCurrentTryCount() >= execQueue.getMaxTryCount()) {
             return false;
         }
-        final long sleepTime = execQueueConfig.sleepTime(execQueue.getCurrentTryCount());
         execQueue.setCurrentTryCount(execQueue.getCurrentTryCount() + 1);
+        final long sleepTime = execQueueConfig.sleepTime(execQueue.getCurrentTryCount());
         execQueue.setNextTime(this.dbHelper.getTime() + sleepTime);
         execQueue.setStatus(ExecQueue.ExecQueueStatus.Wait);
+        this.dbHelper.updateTime(execQueue);
         return true;
     }
 

components/data/MongodbData/src/main/java/com/github/microservice/components/data/mongo/queue/timer/ExecQueueTimer.java

@@ -1,13 +1,21 @@
 package com.github.microservice.components.data.mongo.queue.timer;
 
 
+import com.github.microservice.components.data.mongo.queue.config.ExecQueueConfig;
+import com.github.microservice.components.data.mongo.queue.dao.ExecQueueDao;
+import com.github.microservice.components.data.mongo.queue.domain.ExecQueue;
+import com.github.microservice.components.data.mongo.queue.event.ExecQueueEvent;
+import com.github.microservice.components.data.mongo.queue.service.ExecQueueService;
 import com.github.microservice.components.data.mongo.token.service.ResourceTokenService;
 import com.github.microservice.core.util.os.SystemUtil;
 import lombok.Cleanup;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationEvent;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
@@ -23,6 +31,18 @@ public class ExecQueueTimer implements ApplicationRunner {
     @Autowired
     private ResourceTokenService resourceTokenService;
 
+    @Autowired
+    private ExecQueueDao execQueueDao;
+
+    @Autowired
+    private ExecQueueConfig execQueueConfig;
+
+    @Autowired
+    private ApplicationContext applicationContext;
+
+    @Autowired
+    private ExecQueueService execQueueService;
+
     private final static String RESOURCE_NAME = "ExecQueueTimer";
 
 
@@ -50,7 +70,7 @@ public class ExecQueueTimer implements ApplicationRunner {
         executorService.schedule(() -> {
 
             try {
-                updateWorkStatus();
+                restoreWaitStatus();
             } catch (Exception e) {
                 e.printStackTrace();
             }
@@ -62,19 +82,50 @@ public class ExecQueueTimer implements ApplicationRunner {
                 e.printStackTrace();
             }
 
-
             next();
 
-        }, 1, TimeUnit.SECONDS);
+        }, execQueueConfig.loopEventTime(), TimeUnit.MILLISECONDS);
     }
 
-    // 更新队列的工作状态
-    private void updateWorkStatus() {
-
+    // 恢复状态为等待
+    private void restoreWaitStatus() {
+        long count = this.execQueueDao.restoreWaitStatus(execQueueConfig.executeTimeOut());
+        if (count > 0) {
+            log.info("restoreWaitStatus : {}", count);
+        }
     }
 
     //执行队列
     private void executeWork() {
+        execQueueDao.scanWait(execQueueConfig.maxBatchExecuteCount()).forEach((it) -> {
+            try {
+                execute(it);
+            } catch (Exception e) {
+                e.printStackTrace();
+                log.error(e.getMessage());
+            }
+        });
+    }
+
+    //触发事件
+    @SneakyThrows
+    private void execute(ExecQueue execQueue) {
+        Class<? extends ExecQueueEvent> cls = (Class<? extends ExecQueueEvent>) Class.forName(execQueue.getEventClass());
+        // 发布事件
+        final ExecQueueEvent execQueueEvent = this.applicationContext.getBean(cls);
+
+        //执行任务之前状态修改为工作中
+        this.execQueueDao.updateStatus(execQueue.getId(), ExecQueue.ExecQueueStatus.Work);
+
+        // 修改当前状态为执行完成
+        if (execQueueEvent.execute(execQueue.getData())) {
+            this.execQueueDao.updateStatus(execQueue.getId(), ExecQueue.ExecQueueStatus.Done);
+        } else if (this.execQueueService.updateNextExecTime(execQueue)) {
+            this.execQueueDao.save(execQueue);
+        } else {
+            this.execQueueDao.updateStatus(execQueue.getId(), ExecQueue.ExecQueueStatus.Stop);
+        }
+
 
     }
 

components/data/MongodbData/src/test/java/demo/simple/config/ResourceTokenConfig.java

@@ -1,11 +1,11 @@
-package demo.simple.config;
-
-import com.github.microservice.components.data.mongo.token.config.ResourceTokenConfiguration;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Import;
-
-@Configuration
-@Import(ResourceTokenConfiguration.class)
-public class ResourceTokenConfig {
-
-}
+//package demo.simple.config;
+//
+//import com.github.microservice.components.data.mongo.token.config.ResourceTokenConfiguration;
+//import org.springframework.context.annotation.Configuration;
+//import org.springframework.context.annotation.Import;
+//
+//@Configuration
+//@Import(ResourceTokenConfiguration.class)
+//public class ResourceTokenConfig {
+//
+//}

components/data/MongodbData/src/test/java/demo/simple/config/SimpleTaskTimerConfig.java

@@ -1,30 +1,30 @@
-package demo.simple.config;
-
-import com.github.microservice.components.data.mongo.timer.config.TaskTimerConfiguration;
-import com.github.microservice.components.data.mongo.timer.event.SimpleTaskTimerEvent;
-import demo.simple.domain.TestTaskTimerTable1;
-import demo.simple.domain.TestTaskTimerTable2;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * 任务定时器的配置
- */
-@Configuration
-public class SimpleTaskTimerConfig extends TaskTimerConfiguration {
-
-    @Override
-    public TaskTimerItem[] register() {
-        return new TaskTimerItem[]{
-                TaskTimerItem.builder().taskTimerTable(TestTaskTimerTable1.class).taskTimerEvent((it) -> {
-                    System.out.println(Thread.currentThread() + " TestTaskTimerTable1 -> " + it);
-                }).build()
-                ,
-                new TaskTimerItem().setTaskTimerTable(TestTaskTimerTable2.class).setTaskTimerEvent(new SimpleTaskTimerEvent<TestTaskTimerTable2>() {
-                    @Override
-                    public void execute(TestTaskTimerTable2 it) {
-                        System.out.println(Thread.currentThread() + " TestTaskTimerTable2 -> " + it);
-                    }
-                })
-        };
-    }
-}
+//package demo.simple.config;
+//
+//import com.github.microservice.components.data.mongo.timer.config.TaskTimerConfiguration;
+//import com.github.microservice.components.data.mongo.timer.event.SimpleTaskTimerEvent;
+//import demo.simple.domain.TestTaskTimerTable1;
+//import demo.simple.domain.TestTaskTimerTable2;
+//import org.springframework.context.annotation.Configuration;
+//
+///**
+// * 任务定时器的配置
+// */
+//@Configuration
+//public class SimpleTaskTimerConfig extends TaskTimerConfiguration {
+//
+//    @Override
+//    public TaskTimerItem[] register() {
+//        return new TaskTimerItem[]{
+//                TaskTimerItem.builder().taskTimerTable(TestTaskTimerTable1.class).taskTimerEvent((it) -> {
+//                    System.out.println(Thread.currentThread() + " TestTaskTimerTable1 -> " + it);
+//                }).build()
+//                ,
+//                new TaskTimerItem().setTaskTimerTable(TestTaskTimerTable2.class).setTaskTimerEvent(new SimpleTaskTimerEvent<TestTaskTimerTable2>() {
+//                    @Override
+//                    public void execute(TestTaskTimerTable2 it) {
+//                        System.out.println(Thread.currentThread() + " TestTaskTimerTable2 -> " + it);
+//                    }
+//                })
+//        };
+//    }
+//}

components/data/MongodbData/src/test/java/demo/simple/queue/config/QueueConfig.java

@@ -0,0 +1,23 @@
+package demo.simple.queue.config;
+
+import com.github.microservice.components.data.mongo.queue.config.ExecQueueConfig;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Map;
+
+@Configuration
+public class QueueConfig extends ExecQueueConfig {
+
+    private static final Map<Integer, Long> SleepTime = Map.of(
+            1, 2000L,
+            2, 3000L,
+            3, 5000L,
+            4, 7000L,
+            5, 11000L
+    );
+
+    @Override
+    public long sleepTime(int tryIndex) {
+        return SleepTime.getOrDefault(tryIndex, 1000L * 60);
+    }
+}

components/data/MongodbData/src/test/java/demo/simple/queue/controller/QueueController.java

@@ -0,0 +1,26 @@
+package demo.simple.queue.controller;
+
+import com.github.microservice.components.data.mongo.queue.service.ExecQueueService;
+import demo.simple.queue.event.QuEvent;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Map;
+import java.util.UUID;
+
+@RequestMapping("queue")
+@RestController
+public class QueueController {
+
+    @Autowired
+    private ExecQueueService execQueueService;
+
+    @RequestMapping("add")
+    public Object add(){
+        return execQueueService.add(Map.of("x", UUID.randomUUID()),10, QuEvent.class);
+    }
+
+
+
+}

components/data/MongodbData/src/test/java/demo/simple/queue/event/QuEvent.java

@@ -0,0 +1,27 @@
+package demo.simple.queue.event;
+
+import com.github.microservice.components.data.mongo.queue.event.ExecQueueEvent;
+import com.github.microservice.core.util.random.RandomUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+
+@Slf4j
+@Component
+public class QuEvent implements ExecQueueEvent {
+
+
+    @Override
+    public boolean execute(Map<String,Object> o) {
+        log.info("time : {} -> {}", System.currentTimeMillis(), o);
+        int random = RandomUtil.nextInt(1, 100);
+        if (random % 11 == 0) {
+            return true;
+        }
+//        else if (random % 19 == 0) {
+//            System.exit(0);
+//        }
+        return false;
+    }
+}

components/data/MongodbData/src/test/resources/application.yml

@@ -1,7 +1,7 @@
 spring:
   data:
     mongodb:
-      uri: mongodb://test:test@192.168.8.52:27017,192.168.8.52:27018,192.168.8.52:27019,192.168.8.52:27020/test?replicaSet=MongoSets
+      uri: mongodb://root:8756mongo2024@172.24.50.53:27017,172.24.50.53:27018,172.24.50.53:27019/mongo_jpa_test?authSource=admin
       auto-index-creation: true
 #      database: test
 #      authentication-database: test
@@ -9,8 +9,11 @@ spring:
 #      username: test
 #      password: test
 #      auto-index-creation: true
-
-
+  mvc:
+    pathmatch:
+      matching-strategy: ant_path_matcher
+  main: #允许循环引用
+    allow-circular-references: true
 
 logging:
   level: