feat:自动任务增加redis锁

project-dispatch/pom.xml

@@ -17,6 +17,7 @@
         <druid.version>1.1.10</druid.version>
         <version.org.optaplanner>9.38.0.Final</version.org.optaplanner>
         <mysql.version>8.0.28</mysql.version>
+        <redisson.version>3.19.0</redisson.version>
     </properties>
 
     <dependencies>
@@ -163,6 +164,18 @@
             <version>5.8.22</version>
         </dependency>
 
+        <!-- Redis -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.redisson</groupId>
+            <artifactId>redisson</artifactId>
+            <version>${redisson.version}</version>
+        </dependency>
+
     </dependencies>
     <build>
         <plugins>

project-dispatch/src/main/java/com/dituhui/pea/dispatch/common/RedisConfig.java

+package com.dituhui.pea.dispatch.common;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+@Configuration
+public class RedisConfig {
+
+	@Bean
+	public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory connectionFactory) {
+//		ObjectMapper objectMapper = new ObjectMapper();
+//		objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+//		objectMapper.activateDefaultTyping(objectMapper.getPolymorphicTypeValidator(),
+//				ObjectMapper.DefaultTyping.EVERYTHING, JsonTypeInfo.As.PROPERTY);
+
+		RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
+		redisTemplate.setConnectionFactory(connectionFactory);
+		redisTemplate.setKeySerializer(new StringRedisSerializer());
+		redisTemplate.setValueSerializer(new StringRedisSerializer());
+		return redisTemplate;
+	}
+
+}

project-dispatch/src/main/java/com/dituhui/pea/dispatch/common/RedissonConfig.java

+package com.dituhui.pea.dispatch.common;
+
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.Config;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Objects;
+
+/**
+ * jedis配置类
+ */
+@Configuration
+public class RedissonConfig {
+
+    @Autowired
+    private RedisProperties redisProperties;
+
+    @Bean
+    public RedissonClient redissonClient() {
+        Config config = new Config();
+        if (Objects.isNull(redisProperties.getCluster())
+                || Objects.isNull(redisProperties.getCluster().getNodes())) {
+            //单机配置
+            config.useSingleServer()
+                    .setPassword(redisProperties.getPassword())
+                    .setAddress("redis://" + redisProperties.getHost() + ":" + redisProperties.getPort())
+                    //连接池大小:默认值：64
+                    .setConnectionPoolSize(64);//连接池大小:默认值：64
+        } else {
+            String[] nodes = redisProperties.getCluster().getNodes().toArray(new String[0]);
+            String[] result = new String[nodes.length];
+            for (int i = 0; i < nodes.length; i++) {
+                String nodePart = "redis://" + nodes[i];
+                result[i] = nodePart;
+            }
+            //集群配置
+            config.useClusterServers()
+                    .setPassword(redisProperties.getPassword())
+                    .setScanInterval(10000)
+                    .addNodeAddress(result);
+        }
+
+        return Redisson.create(config);
+    }
+
+}
\ No newline at end of file

project-dispatch/src/main/java/com/dituhui/pea/dispatch/common/RedissonUtil.java

+package com.dituhui.pea.dispatch.common;
+
+import cn.hutool.extra.spring.SpringUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.*;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * TODO redisson 工具类
+ *
+ * @Author RenPing
+ * @Date 2022/1/7 11:04
+ */
+@Slf4j
+@Service
+public class RedissonUtil {
+
+    private static RedissonClient redissonClient;
+
+    private static RedisConnectionFactory redisConnectionFactory;
+
+    @PostConstruct
+    public void init() {
+        redissonClient = SpringUtil.getBean(RedissonClient.class);
+        redisConnectionFactory = SpringUtil.getBean(RedisConnectionFactory.class);
+    }
+
+    /**
+     * 单个加锁操作，示例：lockKey="operate_object_key_1"
+     *
+     * @param lockKey
+     * @param operation
+     */
+    public static void lockOperation(String lockKey, Runnable operation) {
+        RedissonClient redissonClient = SpringUtil.getBean(RedissonClient.class);
+        RLock lock = redissonClient.getLock(lockKey);
+        try {
+            while (true) {
+                //第一个参数：尝试获取锁的最大等待时间，超过这个值，则获取失败
+                //第二个参数：锁的持有时间，超过这个时间锁会自动失效
+                boolean res = lock.tryLock(60, 60, TimeUnit.MINUTES);
+                if (res) {
+                    log.info(">>> " + Thread.currentThread() + " 获取 redis lock");
+                    //成功获得锁，在这里处理业务
+                    operation.run();
+                    break;
+                }
+            }
+        } catch (InterruptedException e) {
+            log.error(e.getMessage(), e);
+            throw new RuntimeException("获取 redis lock 失败");
+        } catch (Exception e) {
+            throw e;
+        } finally {
+            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
+                log.info(">>> " + Thread.currentThread() + " 释放 redis lock");
+                lock.unlock();
+            }
+        }
+    }
+
+}

project-dispatch/src/main/java/com/dituhui/pea/dispatch/service/impl/SchedulerServiceImpl.java

@@ -3,6 +3,7 @@ package com.dituhui.pea.dispatch.service.impl;
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
+import com.dituhui.pea.dispatch.common.RedissonUtil;
 import com.dituhui.pea.dispatch.dao.DispatchBatchRepository;
 import com.dituhui.pea.dispatch.dao.OrgGroupRepository;
 import com.dituhui.pea.dispatch.dao.OrgTeamDao;
@@ -10,6 +11,7 @@ import com.dituhui.pea.dispatch.entity.DispatchBatch;
 import com.dituhui.pea.dispatch.entity.OrgGroup;
 import com.dituhui.pea.dispatch.entity.OrgTeamEntity;
 import com.dituhui.pea.dispatch.pojo.DispatchSolution;
+import com.dituhui.pea.dispatch.quartz.dispatch.AutoDispatchJob;
 import com.dituhui.pea.dispatch.service.BatchService;
 import com.dituhui.pea.dispatch.service.ExtractService;
 import com.dituhui.pea.dispatch.service.SchedulerService;
@@ -22,8 +24,6 @@ import org.optaplanner.persistence.jackson.impl.domain.solution.JacksonSolutionF
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Isolation;
-import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.Resource;
 import java.io.File;
@@ -113,45 +113,52 @@ public class SchedulerServiceImpl implements SchedulerService {
 
         for (int i = 1; i <= nextDaysLimit; i++) {
             String currDay = LocalDate.now().plusDays(i).format(DateTimeFormatter.ISO_LOCAL_DATE);
-            Optional<DispatchBatch> optional = dispatchBatchRepository.findByTeamIdAndBatchDate(teamId, currDay);
-            if (optional.isPresent()
-                    && Objects.nonNull(optional.get().getCutoffedTime())
-                    && today.equals(DateUtil.format(optional.get().getCutoffedTime(), "yyyy-MM-dd"))) {
-                //自动任务截止
-                log.error(">>> teamId:{}, day:{} 自动任务已截止", teamId, currDay);
-                continue;
-            }
+            boolean finalCutOff = cutOff;
+            RedissonUtil.lockOperation(AutoDispatchJob.TEAM_JOB_PREFIX + teamId, () -> {
+                dispatchRun2OneDay(teamId, currDay, today, finalCutOff);
+            });
+        }
+    }
+
+    private void dispatchRun2OneDay(String teamId, String currDay, String today, boolean cutOff) {
+        Optional<DispatchBatch> optional = dispatchBatchRepository.findByTeamIdAndBatchDate(teamId, currDay);
+        if (optional.isPresent()
+                && Objects.nonNull(optional.get().getCutoffedTime())
+                && DateUtil.format(optional.get().getCutoffedTime(), "yyyy-MM-dd").equals(today)) {
+            //自动任务截止
+            log.error(">>> teamId:{}, day:{} 自动任务已截止", teamId, currDay);
+            return;
+        }
+
+        try {
+            log.info("dispatchRun begin----- teamId:{}, day:{}", teamId, currDay);
+
+            String batchNo = batchService.buildBatchData2(teamId, currDay, cutOff);
+            UUID problemId = solveService.generateProblemId(teamId, batchNo);
+            log.info("dispatchRun teamId:{}, day:{}, batch:{}, problemId:{}", teamId, currDay, batchNo, problemId);
+
+            DispatchSolution problem = solveService.prepareSolution2(teamId, batchNo, currDay);
 
-            try {
-                log.info("dispatchRun begin----- teamId:{}, day:{}", teamId, currDay);
-
-                String batchNo = batchService.buildBatchData2(teamId, currDay, cutOff);
-                UUID problemId = solveService.generateProblemId(teamId, batchNo);
-                log.info("dispatchRun teamId:{}, day:{}, batch:{}, problemId:{}", teamId, currDay, batchNo, problemId);
-
-                DispatchSolution problem = solveService.prepareSolution2(teamId, batchNo, currDay);
-
-                if (problem.getCustomerList().size() <= 0) {
-                    log.info("dispatchRun 当前批次没有待指派工单 , teamId:{}, day:{}, batch:{}, problemId:{}, order-size:{}", teamId, currDay, batchNo, problemId, problem.getCustomerList().size());
-                    continue;
-                }
-                log.info("dispatchRun prepare done, teamId:{}, day:{}, batch:{}, problemId:{}", teamId, currDay, batchNo, problemId);
-                Solver<DispatchSolution> solver = solverFactory.buildSolver();
-                DispatchSolution solution = solver.solve(problem);
-                DispatchSolutionUtils.removeHardConstraintCustomer(solution, solverFactory);
-                log.info("dispatchRun solve done, teamId:{}, day:{}, batch:{}, problemId:{}, score:{}", teamId, currDay, batchNo, problemId, solution.getScore().toShortString());
-                this.solveService.saveSolutionWrp2(solution);
-                this.extractService.extractDispatchToOrder2(teamId, batchNo, cutOff);
-                log.info("dispatchRun done ------ teamId:{}, day:{}", teamId, currDay);
-
-                JacksonSolutionFileIO<DispatchSolution> exporter = new JacksonSolutionFileIO<DispatchSolution>(DispatchSolution.class);
-                exporter.write(solution, new File(String.format("dispatchSolution_%s_%s.json", teamId, currDay)));
-
-                //log.info("dispatchRun group:{}, team:{} done", groupId, teamId);
-            } catch (Exception e) {
-                log.error(">>> (teamId:{}, day:{})自动排班失败：{}", teamId, currDay, e.getMessage(), e);
-                throw e;
+            if (problem.getCustomerList().size() <= 0) {
+                log.info("dispatchRun 当前批次没有待指派工单 , teamId:{}, day:{}, batch:{}, problemId:{}, order-size:{}", teamId, currDay, batchNo, problemId, problem.getCustomerList().size());
+                return;
             }
+            log.info("dispatchRun prepare done, teamId:{}, day:{}, batch:{}, problemId:{}", teamId, currDay, batchNo, problemId);
+            Solver<DispatchSolution> solver = solverFactory.buildSolver();
+            DispatchSolution solution = solver.solve(problem);
+            DispatchSolutionUtils.removeHardConstraintCustomer(solution, solverFactory);
+            log.info("dispatchRun solve done, teamId:{}, day:{}, batch:{}, problemId:{}, score:{}", teamId, currDay, batchNo, problemId, solution.getScore().toShortString());
+            this.solveService.saveSolutionWrp2(solution);
+            this.extractService.extractDispatchToOrder2(teamId, batchNo, cutOff);
+            log.info("dispatchRun done ------ teamId:{}, day:{}", teamId, currDay);
+
+            JacksonSolutionFileIO<DispatchSolution> exporter = new JacksonSolutionFileIO<DispatchSolution>(DispatchSolution.class);
+            exporter.write(solution, new File(String.format("dispatchSolution_%s_%s.json", teamId, currDay)));
+
+            //log.info("dispatchRun group:{}, team:{} done", groupId, teamId);
+        } catch (Exception e) {
+            log.error(">>> (teamId:{}, day:{})自动排班失败：{}", teamId, currDay, e.getMessage(), e);
+            //throw e;
         }
     }
 }
\ No newline at end of file

project-dispatch/src/main/resources/application-dev.yaml

@@ -3,9 +3,8 @@ server:
 
 dispatch:
   cron:
-    expr: 0 25 8-23 * * ?
+    expr: 0 */3 8-23 * * ?
     next-day-limit: 20
-#    expr: 0 */10 8-18 * * ?
 
 scheduler:
   init-engineer-capacity:

project-dispatch/src/main/resources/application.yaml

@@ -45,6 +45,9 @@ spring:
         import-check:
           # no config file
           enabled: false
+  config:
+    import:
+      - nacos:redis-config.yaml?group=project&refreshEnabled=true
   datasource:
     driver-class-name: com.mysql.cj.jdbc.Driver
     url: jdbc:mysql://10.10.0.116:32306/saas_aftersale_test?serverTimezone=Asia/Shanghai