feature(*): 项目启动时同步队列

- 项目启动时同步队列

feature(*): 项目启动时同步队列
- 项目启动时同步队列
张志伟
1 parent d8d78651
Showing 8 changed files with 207 additions and 28 deletions
fw-valhalla-rpc/src/main/java/cn/fw/valhalla/rpc/oop/OopService.java
fw-valhalla-rpc/src/main/java/cn/fw/valhalla/rpc/shirasawa/ShirasawaRpcService.java
fw-valhalla-server/src/main/java/cn/fw/valhalla/controller/runner/InitCommandLineRunner.java
fw-valhalla-server/src/main/java/cn/fw/valhalla/controller/task/PubStandTask.java
fw-valhalla-server/src/main/resources/application.yml
fw-valhalla-service/src/main/java/cn/fw/valhalla/service/bus/pub/PubDistributeBizService.java
fw-valhalla-service/src/main/java/cn/fw/valhalla/service/bus/pub/PubStandBizService.java
pom.xml
@@ -15,6 +15,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cache.annotation.Cacheable;
 import org.springframework.stereotype.Service;
 import org.springframework.util.Assert;
  
@@ -225,6 +226,32 @@ public class OopService extends AbsBaseRpcService {
         }
     }
  
+    /**
+     * 查询所有集团
+     *
+     * @return
+     */
+    @Cacheable(cacheNames = "group:all:info", unless = "#result.size() <= 0")
+    public List<GroupDTO> allGroup() {
+        try {
+            Message<List<GroupInfoVo>> msg = groupApiService.getGroupInfoList();
+            List<GroupInfoVo> data = msg.getData();
+            if (msg.isSuccess() && data != null) {
+                List<GroupDTO> list = data.stream().map(item -> {
+                    final GroupDTO groupDTO = new GroupDTO();
+                    BeanUtils.copyProperties(item, groupDTO);
+                    return groupDTO;
+                }).collect(Collectors.toList());
+                if (CollectionUtils.isNotEmpty(list)) {
+                    return list;
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return new ArrayList<>();
+    }
+
     public List<DealerDTO> dealers(final Long groupId) {
         if (groupId == null) {
             return null;
@@ -17,6 +17,7 @@ import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
  
 import java.util.Objects;
+import java.util.Optional;
  
 import static cn.fw.common.businessvalidator.Validator.BV;
  
@@ -107,4 +108,27 @@ public class ShirasawaRpcService {
         final Message<Boolean> msg = followApiService.createNextRecord(newRecordDTO);
         BV.isTrue(msg.isSuccess(), msg::getResult);
     }
+
+    /**
+     * 查询是否有未完成的待办
+     *
+     * @param userId
+     * @return
+     */
+    public boolean hasOngoingFollow(Long userId) {
+        try {
+            if (Objects.isNull(userId)) {
+                return true;
+            }
+            Message<Long> message = followApiService.queryRecordRemaining(userId, BusinessTypeEnum.AS.getValue());
+            log.info("followApiService.queryRecordRemaining: msg.code={}, msg.result={}", message.getCode(), message.getResult());
+            if (message.isSuccess()) {
+                long data = Optional.ofNullable(message.getData()).orElse(0L);
+                return data > 0;
+            }
+        } catch (Exception e) {
+            log.error("查询未完成的待办失败。userId: [{}]", userId, e);
+        }
+        return true;
+    }
 }
+package cn.fw.valhalla.controller.runner;
+
+import cn.fw.valhalla.service.bus.pub.PubStandBizService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.stereotype.Component;
+
+/**
+ * 初始化同步逻辑
+ *
+ * @author : kurisu
+ * @version : 1.0
+ * @className : InitCommandLineRunner
+ * @description : 初始化同步逻辑
+ * @date : 2022-05-09 15:22
+ */
+@Component
+@Slf4j
+@RequiredArgsConstructor
+public class InitCommandLineRunner implements CommandLineRunner {
+    private final PubStandBizService pubStandBizService;
+
+    @Override
+    public void run(String... args) {
+        log.info("项目启动成功，开始同步公共池站岗队列");
+        pubStandBizService.syncQueue();
+    }
+}
+package cn.fw.valhalla.controller.task;
+
+import cn.fw.common.cache.locker.DistributedLocker;
+import cn.fw.valhalla.common.utils.StringUtils;
+import cn.fw.valhalla.rpc.oop.OopService;
+import cn.fw.valhalla.rpc.oop.dto.GroupDTO;
+import cn.fw.valhalla.service.bus.pub.PubDistributeBizService;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
+import org.redisson.api.RLock;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.redis.core.BoundListOperations;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+import java.time.MonthDay;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 公共池站岗任务
+ *
+ * @author : kurisu
+ * @version : 1.0
+ * @className : PubStandTask
+ * @description : 公共池站岗任务
+ * @date : 2023-03-15 11:01
+ */
+@Component
+@ConditionalOnProperty(prefix = "task", name = "switch", havingValue = "on")
+@Slf4j
+@RequiredArgsConstructor
+public class PubStandTask {
+    private final PubDistributeBizService pubDistributeBizService;
+    private final StringRedisTemplate redisTemplate;
+    private final OopService oopService;
+    private final DistributedLocker distributedLocker;
+    @Value("${spring.cache.custom.global-prefix}:stand:pub")
+    @Getter(AccessLevel.PRIVATE)
+    private String keyPrefix;
+
+    /**
+     * 执行分配
+     */
+    @Scheduled(initialDelay = 1000 * 10, fixedRate = 1000 * 5)
+    public void distributeBatch() {
+        List<GroupDTO> groups = oopService.allGroup();
+        for (GroupDTO group : groups) {
+            CompletableFuture.runAsync(() -> doDistribute(group));
+        }
+    }
+
+    private void doDistribute(GroupDTO group) {
+        final String key = generateKey(group.getId());
+        final String lockKey = String.format("pub:distribute:%s", group.getId());
+        Pair<Boolean, RLock> lockPair = distributedLocker.tryLock(lockKey, TimeUnit.MINUTES, 0, -1);
+        try {
+            if (Boolean.TRUE.equals(lockPair.getLeft())) {
+                List<String> failList = new ArrayList<>();
+                BoundListOperations<String, String> ops = redisTemplate.boundListOps(key);
+                String id;
+                while ((id = ops.leftPop()) != null) {
+                    if (StringUtils.isNumber(id)) {
+                        try {
+                            boolean distributed = pubDistributeBizService.distribute(Long.valueOf(id));
+                            if (!distributed) {
+                                failList.add(id);
+                            }
+                        } catch (Exception e) {
+                            if (StringUtils.isValid(id)) {
+                                failList.add(id);
+                            }
+                            log.error("分配公共池客户", e);
+                        }
+                    }
+                }
+                if (!CollectionUtils.isEmpty(failList)) {
+                    ops.rightPushAll(failList.toArray(new String[0]));
+                }
+            }
+        } finally {
+            if (lockPair.getRight().isLocked()) {
+                lockPair.getRight().unlock();
+            }
+        }
+    }
+
+    private String generateKey(final Long groupId) {
+        Assert.notNull(groupId, "groupId cannot be null");
+        String day = MonthDay.now().toString().replace("-", "");
+        return String.format("%s:%s:%s", getKeyPrefix(), day, groupId);
+    }
+}
@@ -9,12 +9,14 @@ spring:
     custom:
       global-prefix: 'valhalla'
       global:
-        ttl: 12h
+        ttl: 10m
       cache:
         - name: mobile:attribution
           ttl: 8h
         - name: pub:stand
           ttl: 12h
+        - name: group:all:info
+          ttl: 2h
  
   cloud:
     nacos:
@@ -2,22 +2,19 @@ package cn.fw.valhalla.service.bus.pub;
  
 import cn.fw.common.cache.CacheExtender;
 import cn.fw.common.cache.locker.DistributedLocker;
+import cn.fw.valhalla.domain.db.pub.PubStandStaffInfo;
 import cn.fw.valhalla.rpc.ehr.EhrRpcService;
 import cn.fw.valhalla.service.data.PubStandStaffInfoService;
 import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.tuple.Pair;
-import org.redisson.api.RLock;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.data.redis.core.BoundListOperations;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.util.Assert;
  
 import java.time.MonthDay;
-import java.util.concurrent.TimeUnit;
  
 /**
  * 公共池站岗分配
@@ -40,26 +37,16 @@ public class PubDistributeBizService {
     @Value("${spring.cache.custom.global-prefix}:stand:pub")
     @Getter(AccessLevel.PRIVATE)
     private String keyPrefix;
+
     /**
      * 执行分配
      *
-     * @param groupId
+     * @param id
      */
-    public void distribute(Long groupId) {
-        final String key = generateKey(groupId);
-        final String lockKey = String.format("pub:distribute:%s", groupId);
-        Pair<Boolean, RLock> lockPair = distributedLocker.tryLock(lockKey, TimeUnit.MINUTES, 0, -1);
-        try {
-            if (Boolean.TRUE.equals(lockPair.getLeft())) {
-                BoundListOperations<String, String> ops = redisTemplate.boundListOps(key);
-                String id = ops.leftPop();
-            }
-        } finally {
-            if (lockPair.getRight().isLocked()) {
-                lockPair.getRight().unlock();
-            }
-        }
-
+    public boolean distribute(Long id) {
+        PubStandStaffInfo staffInfo = pubStandStaffInfoService.getById(id);
+        // TODO: 2023/3/15 分配
+        return true;
     }
  
     private String generateKey(final Long groupId) {
 package cn.fw.valhalla.service.bus.pub;
  
-import cn.fw.common.cache.CacheExtender;
-import cn.fw.common.cache.locker.DistributedLocker;
 import cn.fw.common.web.annotation.DisLock;
 import cn.fw.erp.sdk.api.result.UserRoleDataRange;
 import cn.fw.valhalla.common.constant.RoleCode;
@@ -13,7 +11,9 @@ import cn.fw.valhalla.rpc.ehr.EhrRpcService;
 import cn.fw.valhalla.rpc.erp.UserRoleRpcService;
 import cn.fw.valhalla.rpc.erp.dto.UserInfoDTO;
 import cn.fw.valhalla.rpc.oop.OopService;
+import cn.fw.valhalla.rpc.oop.dto.GroupDTO;
 import cn.fw.valhalla.rpc.oop.dto.ShopDTO;
+import cn.fw.valhalla.rpc.shirasawa.ShirasawaRpcService;
 import cn.fw.valhalla.service.data.PubStandStaffInfoService;
 import lombok.AccessLevel;
 import lombok.Getter;
@@ -54,9 +54,8 @@ public class PubStandBizService {
     private final OopService oopService;
     private final UserRoleRpcService userRoleRpcService;
     private final PubStandStaffInfoService pubStandStaffInfoService;
-    private final CacheExtender cache;
+    private final ShirasawaRpcService shirasawaRpcService;
     private final StringRedisTemplate redisTemplate;
-    private final DistributedLocker distributedLocker;
     private final PlatformTransactionManager platformTransactionManager;
     private final TransactionDefinition transactionDefinition;
     @Value("${spring.cache.custom.global-prefix}:stand:pub")
@@ -65,6 +64,16 @@ public class PubStandBizService {
  
     /**
      * 同步站岗队列
+     */
+    public void syncQueue() {
+        List<GroupDTO> list = oopService.allGroup();
+        for (GroupDTO group : list) {
+            syncQueue(group.getId());
+        }
+    }
+
+    /**
+     * 同步站岗队列
      *
      * @param groupId
      */
@@ -184,8 +193,7 @@ public class PubStandBizService {
     }
  
     private boolean hasAnyTodo(final Long staffId) {
-        //todo 跟进系统 跟进结果返回生于的跟进数量并且提供主动查询的接口
-        return true;
+        return shirasawaRpcService.hasOngoingFollow(staffId);
     }
  
     private PubStandStaffInfo generateStandInfo(UserInfoDTO infoDTO) {
@@ -64,7 +64,7 @@
         <yitter.idgenerator>1.0.3</yitter.idgenerator>
         <fw.hestia.sdk>1.0.0</fw.hestia.sdk>
         <fw.backlog.sdk>1.0.0</fw.backlog.sdk>
-        <fw.shirasawa.sdk>1.0.2</fw.shirasawa.sdk>
+        <fw.shirasawa.sdk>1.0.3</fw.shirasawa.sdk>
         <fw-identify-sdk.version>1.0.0</fw-identify-sdk.version>
         <fw-attendance-sdk.version>1.0.0</fw-attendance-sdk.version>
     </properties>