优化 mp 账号的刷新机制，使用 Job 轮询，替换 MQ 广播

YunaiV
1 parent 5d6308e4
Showing 14 changed files with 62 additions and 115 deletions
yudao-framework/yudao-common/src/main/java/cn/iocoder/yudao/framework/common/util/collection/CollectionUtils.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/service/definition/BpmProcessInstanceServiceImpl.java
yudao-module-bpm/yudao-module-bpm-biz/src/main/java/cn/iocoder/yudao/module/bpm/service/definition/BpmTaskAssignRuleServiceImpl.java
yudao-module-bpm/yudao-module-bpm-biz/src/test/java/cn/iocoder/yudao/module/bpm/framework/flowable/core/behavior/script/impl/BpmTaskAssignLeaderX2ScriptTest.java
yudao-module-mp/yudao-module-mp-biz/src/main/java/cn/iocoder/yudao/module/mp/dal/mysql/account/MpAccountMapper.java
yudao-module-mp/yudao-module-mp-biz/src/main/java/cn/iocoder/yudao/module/mp/mq/consumer/MpAccountRefreshConsumer.java
yudao-module-mp/yudao-module-mp-biz/src/main/java/cn/iocoder/yudao/module/mp/mq/message/MpAccountRefreshMessage.java
yudao-module-mp/yudao-module-mp-biz/src/main/java/cn/iocoder/yudao/module/mp/mq/producer/MpAccountProducer.java
yudao-module-mp/yudao-module-mp-biz/src/main/java/cn/iocoder/yudao/module/mp/service/account/MpAccountServiceImpl.java
yudao-module-pay/yudao-module-pay-api/src/main/java/cn/iocoder/yudao/module/pay/api/order/dto/PayOrderRespDTO.java
yudao-module-pay/yudao-module-pay-api/src/main/java/cn/iocoder/yudao/module/pay/api/refund/dto/PayRefundRespDTO.java
yudao-module-pay/yudao-module-pay-biz/pom.xml
yudao-module-pay/yudao-module-pay-biz/src/main/java/cn/iocoder/yudao/module/pay/service/channel/PayChannelServiceImpl.java
yudao-server/src/main/resources/application-dev.yaml
@@ -164,7 +164,7 @@ public class CollectionUtils {
         return from.stream().filter(predicate).findFirst().orElse(null);
     }
-    public static <T, V extends Comparable<? super V>> V getMaxValue(List<T> from, Function<T, V> valueFunc) {
+    public static <T, V extends Comparable<? super V>> V getMaxValue(Collection<T> from, Function<T, V> valueFunc) {
         if (CollUtil.isEmpty(from)) {
             return null;
         }
+package cn.iocoder.yudao.module.bpm.service.definition;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.framework.common.pojo.PageResult;
import cn.iocoder.yudao.framework.common.util.number.NumberUtils;
import cn.iocoder.yudao.module.bpm.api.task.dto.BpmProcessInstanceCreateReqDTO;
import cn.iocoder.yudao.module.bpm.controller.admin.task.vo.instance.*;
import cn.iocoder.yudao.module.bpm.convert.task.BpmProcessInstanceConvert;
import cn.iocoder.yudao.module.bpm.dal.dataobject.definition.BpmProcessDefinitionExtDO;
import cn.iocoder.yudao.module.bpm.dal.dataobject.task.BpmProcessInstanceExtDO;
import cn.iocoder.yudao.module.bpm.dal.mysql.task.BpmProcessInstanceExtMapper;
import cn.iocoder.yudao.module.bpm.enums.task.BpmProcessInstanceDeleteReasonEnum;
import cn.iocoder.yudao.module.bpm.enums.task.BpmProcessInstanceResultEnum;
import cn.iocoder.yudao.module.bpm.enums.task.BpmProcessInstanceStatusEnum;
import cn.iocoder.yudao.module.bpm.framework.bpm.core.event.BpmProcessInstanceResultEventPublisher;
import cn.iocoder.yudao.module.bpm.service.message.BpmMessageService;
import cn.iocoder.yudao.module.bpm.service.task.BpmProcessInstanceService;
import cn.iocoder.yudao.module.bpm.service.task.BpmTaskService;
import cn.iocoder.yudao.module.system.api.dept.DeptApi;
import cn.iocoder.yudao.module.system.api.dept.dto.DeptRespDTO;
import cn.iocoder.yudao.module.system.api.user.AdminUserApi;
import cn.iocoder.yudao.module.system.api.user.dto.AdminUserRespDTO;
import lombok.extern.slf4j.Slf4j;
import org.flowable.engine.HistoryService;
import org.flowable.engine.RuntimeService;
import org.flowable.engine.delegate.event.FlowableCancelledEvent;
import org.flowable.engine.history.HistoricProcessInstance;
import org.flowable.engine.repository.ProcessDefinition;
import org.flowable.engine.runtime.ProcessInstance;
import org.flowable.task.api.Task;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.validation.annotation.Validated;

import javax.annotation.Resource;
import javax.validation.Valid;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
import static cn.iocoder.yudao.module.bpm.enums.ErrorCodeConstants.*;

/**
 * 流程实例 Service 实现类
 *
 * ProcessDefinition & ProcessInstance & Execution & Task 的关系：
 *     1. <a href="https://blog.csdn.net/bobozai86/article/details/105210414" />
 *
 * HistoricProcessInstance & ProcessInstance 的关系：
 *     1. <a href=" https://my.oschina.net/843294669/blog/71902" />
 *
 * 简单来说，前者 = 历史 + 运行中的流程实例，后者仅是运行中的流程实例
 *
 * @author 芋道源码
 */
@Service
@Validated
@Slf4j
public class BpmProcessInstanceServiceImpl implements BpmProcessInstanceService {

    @Resource
    private RuntimeService runtimeService;
    @Resource
    private BpmProcessInstanceExtMapper processInstanceExtMapper;
    @Resource
    @Lazy // 解决循环依赖
    private BpmTaskService taskService;
    @Resource
    private BpmProcessDefinitionService processDefinitionService;
    @Resource
    private HistoryService historyService;
    @Resource
    private AdminUserApi adminUserApi;
    @Resource
    private DeptApi deptApi;
    @Resource
    private BpmProcessInstanceResultEventPublisher processInstanceResultEventPublisher;
    @Resource
    private BpmMessageService messageService;
    @Override
    public ProcessInstance getProcessInstance(String id) {
        return runtimeService.createProcessInstanceQuery().processInstanceId(id).singleResult();
    }

    @Override
    public List<ProcessInstance> getProcessInstances(Set<String> ids) {
        return runtimeService.createProcessInstanceQuery().processInstanceIds(ids).list();
    }

    @Override
    public PageResult<BpmProcessInstancePageItemRespVO> getMyProcessInstancePage(Long userId,
                                                                                 BpmProcessInstanceMyPageReqVO pageReqVO) {
        // 通过 BpmProcessInstanceExtDO 表，先查询到对应的分页
        PageResult<BpmProcessInstanceExtDO> pageResult = processInstanceExtMapper.selectPage(userId, pageReqVO);
        if (CollUtil.isEmpty(pageResult.getList())) {
            return new PageResult<>(pageResult.getTotal());
        }

        // 获得流程 Task Map
        List<String> processInstanceIds = convertList(pageResult.getList(), BpmProcessInstanceExtDO::getProcessInstanceId);
        Map<String, List<Task>> taskMap = taskService.getTaskMapByProcessInstanceIds(processInstanceIds);
        // 转换返回
        return BpmProcessInstanceConvert.INSTANCE.convertPage(pageResult, taskMap);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public String createProcessInstance(Long userId, @Valid BpmProcessInstanceCreateReqVO createReqVO) {
        // 获得流程定义
        ProcessDefinition definition = processDefinitionService.getProcessDefinition(createReqVO.getProcessDefinitionId());
        // 发起流程
        return createProcessInstance0(userId, definition, createReqVO.getVariables(), null);
    }

    @Override
    public String createProcessInstance(Long userId, @Valid BpmProcessInstanceCreateReqDTO createReqDTO) {
        // 获得流程定义
        ProcessDefinition definition = processDefinitionService.getActiveProcessDefinition(createReqDTO.getProcessDefinitionKey());
        // 发起流程
        return createProcessInstance0(userId, definition, createReqDTO.getVariables(), createReqDTO.getBusinessKey());
    }

    @Override
    public BpmProcessInstanceRespVO getProcessInstanceVO(String id) {
        // 获得流程实例
        HistoricProcessInstance processInstance = getHistoricProcessInstance(id);
        if (processInstance == null) {
            return null;
        }
        BpmProcessInstanceExtDO processInstanceExt = processInstanceExtMapper.selectByProcessInstanceId(id);
        Assert.notNull(processInstanceExt, "流程实例拓展({}) 不存在", id);

        // 获得流程定义
        ProcessDefinition processDefinition = processDefinitionService
                .getProcessDefinition(processInstance.getProcessDefinitionId());
        Assert.notNull(processDefinition, "流程定义({}) 不存在", processInstance.getProcessDefinitionId());
        BpmProcessDefinitionExtDO processDefinitionExt = processDefinitionService.getProcessDefinitionExt(
                processInstance.getProcessDefinitionId());
        Assert.notNull(processDefinitionExt, "流程定义拓展({}) 不存在", id);
        String bpmnXml = processDefinitionService.getProcessDefinitionBpmnXML(processInstance.getProcessDefinitionId());

        // 获得 User
        AdminUserRespDTO startUser = adminUserApi.getUser(NumberUtils.parseLong(processInstance.getStartUserId()));
        DeptRespDTO dept = null;
        if (startUser != null) {
            dept = deptApi.getDept(startUser.getDeptId());
        }

        // 拼接结果
        return BpmProcessInstanceConvert.INSTANCE.convert2(processInstance, processInstanceExt,
                processDefinition, processDefinitionExt, bpmnXml, startUser, dept);
    }

    @Override
    public void cancelProcessInstance(Long userId, @Valid BpmProcessInstanceCancelReqVO cancelReqVO) {
        // 校验流程实例存在
        ProcessInstance instance = getProcessInstance(cancelReqVO.getId());
        if (instance == null) {
            throw exception(PROCESS_INSTANCE_CANCEL_FAIL_NOT_EXISTS);
        }
        // 只能取消自己的
        if (!Objects.equals(instance.getStartUserId(), String.valueOf(userId))) {
            throw exception(PROCESS_INSTANCE_CANCEL_FAIL_NOT_SELF);
        }

        // 通过删除流程实例，实现流程实例的取消,
        // 删除流程实例，正则执行任务 ACT_RU_TASK. 任务会被删除。通过历史表查询
        deleteProcessInstance(cancelReqVO.getId(),
                BpmProcessInstanceDeleteReasonEnum.CANCEL_TASK.format(cancelReqVO.getReason()));
    }

    /**
     * 获得历史的流程实例
     *
     * @param id 流程实例的编号
     * @return 历史的流程实例
     */
    @Override
    public HistoricProcessInstance getHistoricProcessInstance(String id) {
        return historyService.createHistoricProcessInstanceQuery().processInstanceId(id).singleResult();
    }

    @Override
    public List<HistoricProcessInstance> getHistoricProcessInstances(Set<String> ids) {
        return historyService.createHistoricProcessInstanceQuery().processInstanceIds(ids).list();
    }

    @Override
    public void createProcessInstanceExt(ProcessInstance instance) {
        // 获得流程定义
        ProcessDefinition definition = processDefinitionService.getProcessDefinition2(instance.getProcessDefinitionId());
        // 插入 BpmProcessInstanceExtDO 对象
        BpmProcessInstanceExtDO instanceExtDO = new BpmProcessInstanceExtDO()
                .setProcessInstanceId(instance.getId())
                .setProcessDefinitionId(definition.getId())
                .setName(instance.getProcessDefinitionName())
                .setStartUserId(Long.valueOf(instance.getStartUserId()))
                .setCategory(definition.getCategory())
                .setStatus(BpmProcessInstanceStatusEnum.RUNNING.getStatus())
                .setResult(BpmProcessInstanceResultEnum.PROCESS.getResult());

        processInstanceExtMapper.insert(instanceExtDO);
    }

    @Override
    public void updateProcessInstanceExtCancel(FlowableCancelledEvent event) {
        // 判断是否为 Reject 不通过。如果是，则不进行更新.
        // 因为，updateProcessInstanceExtReject 方法，已经进行更新了
        if (BpmProcessInstanceDeleteReasonEnum.isRejectReason((String)event.getCause())) {
            return;
        }

        // 需要主动查询，因为 instance 只有 id 属性
        // 另外，此时如果去查询 ProcessInstance 的话，字段是不全的，所以去查询了 HistoricProcessInstance
        HistoricProcessInstance processInstance = getHistoricProcessInstance(event.getProcessInstanceId());
        // 更新拓展表
        BpmProcessInstanceExtDO instanceExtDO = new BpmProcessInstanceExtDO()
                .setProcessInstanceId(event.getProcessInstanceId())
                .setEndTime(LocalDateTime.now()) // 由于 ProcessInstance 里没有办法拿到 endTime，所以这里设置
                .setStatus(BpmProcessInstanceStatusEnum.FINISH.getStatus())
                .setResult(BpmProcessInstanceResultEnum.CANCEL.getResult());
        processInstanceExtMapper.updateByProcessInstanceId(instanceExtDO);

        // 发送流程实例的状态事件
        processInstanceResultEventPublisher.sendProcessInstanceResultEvent(
                BpmProcessInstanceConvert.INSTANCE.convert(this, processInstance, instanceExtDO.getResult()));
    }

    @Override
    public void updateProcessInstanceExtComplete(ProcessInstance instance) {
        // 需要主动查询，因为 instance 只有 id 属性
        // 另外，此时如果去查询 ProcessInstance 的话，字段是不全的，所以去查询了 HistoricProcessInstance
        HistoricProcessInstance processInstance = getHistoricProcessInstance(instance.getId());
        // 更新拓展表
        BpmProcessInstanceExtDO instanceExtDO = new BpmProcessInstanceExtDO()
                .setProcessInstanceId(instance.getProcessInstanceId())
                .setEndTime(LocalDateTime.now()) // 由于 ProcessInstance 里没有办法拿到 endTime，所以这里设置
                .setStatus(BpmProcessInstanceStatusEnum.FINISH.getStatus())
                .setResult(BpmProcessInstanceResultEnum.APPROVE.getResult()); // 如果正常完全，说明审批通过
        processInstanceExtMapper.updateByProcessInstanceId(instanceExtDO);

        // 发送流程被通过的消息
        messageService.sendMessageWhenProcessInstanceApprove(BpmProcessInstanceConvert.INSTANCE.convert2ApprovedReq(instance));

        // 发送流程实例的状态事件
        processInstanceResultEventPublisher.sendProcessInstanceResultEvent(
                BpmProcessInstanceConvert.INSTANCE.convert(this, processInstance, instanceExtDO.getResult()));
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateProcessInstanceExtReject(String id, String reason) {
        // 需要主动查询，因为 instance 只有 id 属性
        ProcessInstance processInstance = getProcessInstance(id);
        // 删除流程实例，以实现驳回任务时，取消整个审批流程
        deleteProcessInstance(id, StrUtil.format(BpmProcessInstanceDeleteReasonEnum.REJECT_TASK.format(reason)));

        // 更新 status + result
        // 注意，不能和上面的逻辑更换位置。因为 deleteProcessInstance 会触发流程的取消，进而调用 updateProcessInstanceExtCancel 方法，
        // 设置 result 为 BpmProcessInstanceStatusEnum.CANCEL，显然和 result 不一定是一致的
        BpmProcessInstanceExtDO instanceExtDO = new BpmProcessInstanceExtDO().setProcessInstanceId(id)
                .setStatus(BpmProcessInstanceStatusEnum.FINISH.getStatus())
                .setResult(BpmProcessInstanceResultEnum.REJECT.getResult());
        processInstanceExtMapper.updateByProcessInstanceId(instanceExtDO);

        // 发送流程被不通过的消息
        messageService.sendMessageWhenProcessInstanceReject(BpmProcessInstanceConvert.INSTANCE.convert2RejectReq(processInstance, reason));

        // 发送流程实例的状态事件
        processInstanceResultEventPublisher.sendProcessInstanceResultEvent(
                BpmProcessInstanceConvert.INSTANCE.convert(this, processInstance, instanceExtDO.getResult()));
    }

    private void deleteProcessInstance(String id, String reason) {
        runtimeService.deleteProcessInstance(id, reason);
    }

    private String createProcessInstance0(Long userId, ProcessDefinition definition,
                                          Map<String, Object> variables, String businessKey) {
        // 校验流程定义
        if (definition == null) {
            throw exception(PROCESS_DEFINITION_NOT_EXISTS);
        }
        if (definition.isSuspended()) {
            throw exception(PROCESS_DEFINITION_IS_SUSPENDED);
        }

        // 创建流程实例
        ProcessInstance instance = runtimeService.createProcessInstanceBuilder()
                .processDefinitionId(definition.getId())
                .businessKey(businessKey)
                .name(definition.getName().trim())
                .variables(variables)
                .start();
        // 设置流程名字
        runtimeService.setProcessInstanceName(instance.getId(), definition.getName());

        // 补全流程实例的拓展表
        processInstanceExtMapper.updateByProcessInstanceId(new BpmProcessInstanceExtDO().setProcessInstanceId(instance.getId())
                .setFormVariables(variables));
        return instance.getId();
    }

}
 \ No newline at end of file
@@ -15,8 +15,8 @@ import cn.iocoder.yudao.module.bpm.convert.definition.BpmTaskAssignRuleConvert;
 import cn.iocoder.yudao.module.bpm.dal.dataobject.definition.BpmTaskAssignRuleDO;
 import cn.iocoder.yudao.module.bpm.dal.dataobject.definition.BpmUserGroupDO;
 import cn.iocoder.yudao.module.bpm.dal.mysql.definition.BpmTaskAssignRuleMapper;
-import cn.iocoder.yudao.module.bpm.enums.definition.BpmTaskAssignRuleTypeEnum;
 import cn.iocoder.yudao.module.bpm.enums.DictTypeConstants;
+import cn.iocoder.yudao.module.bpm.enums.definition.BpmTaskAssignRuleTypeEnum;
 import cn.iocoder.yudao.module.bpm.framework.flowable.core.behavior.script.BpmTaskAssignScript;
 import cn.iocoder.yudao.module.system.api.dept.DeptApi;
 import cn.iocoder.yudao.module.system.api.dept.PostApi;
@@ -89,7 +89,7 @@ public class BpmTaskAssignRuleServiceImpl implements BpmTaskAssignRuleService {
     @Override
     public List<BpmTaskAssignRuleDO> getTaskAssignRuleListByProcessDefinitionId(String processDefinitionId,
-        String taskDefinitionKey) {
+                                                                                String taskDefinitionKey) {
         return taskRuleMapper.selectListByProcessDefinitionId(processDefinitionId, taskDefinitionKey);
     }
@@ -128,14 +128,14 @@ public class BpmTaskAssignRuleServiceImpl implements BpmTaskAssignRuleService {
         validTaskAssignRuleOptions(reqVO.getType(), reqVO.getOptions());
         // 校验是否已经配置
         BpmTaskAssignRuleDO existRule =
-            taskRuleMapper.selectListByModelIdAndTaskDefinitionKey(reqVO.getModelId(), reqVO.getTaskDefinitionKey());
+                taskRuleMapper.selectListByModelIdAndTaskDefinitionKey(reqVO.getModelId(), reqVO.getTaskDefinitionKey());
         if (existRule != null) {
             throw exception(TASK_ASSIGN_RULE_EXISTS, reqVO.getModelId(), reqVO.getTaskDefinitionKey());
         }
         // 存储
         BpmTaskAssignRuleDO rule = BpmTaskAssignRuleConvert.INSTANCE.convert(reqVO)
-            .setProcessDefinitionId(BpmTaskAssignRuleDO.PROCESS_DEFINITION_ID_NULL); // 只有流程模型，才允许新建
+                .setProcessDefinitionId(BpmTaskAssignRuleDO.PROCESS_DEFINITION_ID_NULL); // 只有流程模型，才允许新建
         taskRuleMapper.insert(rule);
         return rule.getId();
     }
@@ -169,14 +169,14 @@ public class BpmTaskAssignRuleServiceImpl implements BpmTaskAssignRuleService {
         // 遍历，匹配对应的规则
         Map<String, BpmTaskAssignRuleRespVO> processInstanceRuleMap =
-            CollectionUtils.convertMap(processInstanceRules, BpmTaskAssignRuleRespVO::getTaskDefinitionKey);
+                CollectionUtils.convertMap(processInstanceRules, BpmTaskAssignRuleRespVO::getTaskDefinitionKey);
         for (BpmTaskAssignRuleRespVO modelRule : modelRules) {
             BpmTaskAssignRuleRespVO processInstanceRule = processInstanceRuleMap.get(modelRule.getTaskDefinitionKey());
             if (processInstanceRule == null) {
                 return false;
             }
             if (!ObjectUtil.equals(modelRule.getType(), processInstanceRule.getType()) || !ObjectUtil.equal(
-                modelRule.getOptions(), processInstanceRule.getOptions())) {
+                    modelRule.getOptions(), processInstanceRule.getOptions())) {
                 return false;
             }
         }
@@ -192,7 +192,7 @@ public class BpmTaskAssignRuleServiceImpl implements BpmTaskAssignRuleService {
         // 开始复制
         List<BpmTaskAssignRuleDO> newRules = BpmTaskAssignRuleConvert.INSTANCE.convertList2(rules);
         newRules.forEach(rule -> rule.setProcessDefinitionId(toProcessDefinitionId).setId(null).setCreateTime(null)
-            .setUpdateTime(null));
+                .setUpdateTime(null));
         taskRuleMapper.insertBatch(newRules);
     }
@@ -215,7 +215,7 @@ public class BpmTaskAssignRuleServiceImpl implements BpmTaskAssignRuleService {
         if (Objects.equals(type, BpmTaskAssignRuleTypeEnum.ROLE.getType())) {
             roleApi.validRoleList(options);
         } else if (ObjectUtils.equalsAny(type, BpmTaskAssignRuleTypeEnum.DEPT_MEMBER.getType(),
-            BpmTaskAssignRuleTypeEnum.DEPT_LEADER.getType())) {
+                BpmTaskAssignRuleTypeEnum.DEPT_LEADER.getType())) {
             deptApi.validateDeptList(options);
         } else if (Objects.equals(type, BpmTaskAssignRuleTypeEnum.POST.getType())) {
             postApi.validPostList(options);
@@ -225,7 +225,7 @@ public class BpmTaskAssignRuleServiceImpl implements BpmTaskAssignRuleService {
             userGroupService.validUserGroups(options);
         } else if (Objects.equals(type, BpmTaskAssignRuleTypeEnum.SCRIPT.getType())) {
             dictDataApi.validateDictDataList(DictTypeConstants.TASK_ASSIGN_SCRIPT,
-                CollectionUtils.convertSet(options, String::valueOf));
+                    CollectionUtils.convertSet(options, String::valueOf));
         } else {
             throw new IllegalArgumentException(format("未知的规则类型({})", type));
         }
@@ -40,6 +40,8 @@ public class BpmTaskAssignLeaderX2ScriptTest extends BaseMockitoUnitTest {
         // mock 方法(startUser)
         AdminUserRespDTO startUser = randomPojo(AdminUserRespDTO.class, o -> o.setDeptId(10L));
         when(adminUserApi.getUser(eq(1L))).thenReturn(startUser);
+        // mock 方法(getStartUserDept)没有部门
+        when(deptApi.getDept(eq(10L))).thenReturn(null);
         // 调用
         Set<Long> result = script.calculateTaskCandidateUsers(execution);
@@ -56,7 +58,9 @@ public class BpmTaskAssignLeaderX2ScriptTest extends BaseMockitoUnitTest {
         when(adminUserApi.getUser(eq(1L))).thenReturn(startUser);
         DeptRespDTO startUserDept = randomPojo(DeptRespDTO.class, o -> o.setId(10L).setParentId(100L)
                 .setLeaderUserId(20L));
+        // mock 方法（getDept）
         when(deptApi.getDept(eq(10L))).thenReturn(startUserDept);
+        when(deptApi.getDept(eq(100L))).thenReturn(null);
         // 调用
         Set<Long> result = script.calculateTaskCandidateUsers(execution);
@@ -6,6 +6,9 @@ import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
 import cn.iocoder.yudao.module.mp.controller.admin.account.vo.MpAccountPageReqVO;
 import cn.iocoder.yudao.module.mp.dal.dataobject.account.MpAccountDO;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Select;
+
+import java.time.LocalDateTime;
 @Mapper
 public interface MpAccountMapper extends BaseMapperX<MpAccountDO> {
@@ -22,4 +25,7 @@ public interface MpAccountMapper extends BaseMapperX&lt;MpAccountDO&gt; {
         return selectOne(MpAccountDO::getAppId, appId);
     }
+    @Select("SELECT COUNT(*) FROM pay_account WHERE update_time > #{maxUpdateTime}")
+    Long selectCountByUpdateTimeGt(LocalDateTime maxUpdateTime);
+
 }
-package cn.iocoder.yudao.module.mp.mq.consumer;
-
-import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener;
-import cn.iocoder.yudao.module.mp.mq.message.MpAccountRefreshMessage;
-import cn.iocoder.yudao.module.mp.service.account.MpAccountService;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-
-import javax.annotation.Resource;
-
-/**
- * 针对 {@link MpAccountRefreshMessage} 的消费者
- *
- * @author 芋道源码
- */
-@Component
-@Slf4j
-public class MpAccountRefreshConsumer extends AbstractChannelMessageListener<MpAccountRefreshMessage> {
-
-    @Resource
-    private MpAccountService mpAccountService;
-
-    @Override
-    public void onMessage(MpAccountRefreshMessage message) {
-        log.info("[onMessage][收到 Account 刷新消息]");
-        mpAccountService.initLocalCache();
-    }
-
-}
-package cn.iocoder.yudao.module.mp.mq.message;
-
-import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessage;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-
-/**
- * 公众号账号刷新 Message
- *
- * @author 芋道源码
- */
-@Data
-@EqualsAndHashCode(callSuper = true)
-public class MpAccountRefreshMessage extends AbstractChannelMessage {
-
-    @Override
-    public String getChannel() {
-        return "mp.account.refresh";
-    }
-
-}
-package cn.iocoder.yudao.module.mp.mq.producer;
-
-import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate;
-import cn.iocoder.yudao.module.mp.mq.message.MpAccountRefreshMessage;
-import org.springframework.stereotype.Component;
-
-import javax.annotation.Resource;
-
-/**
- * 公众号账号 Producer
- *
- * @author 芋道源码
- */
-@Component
-public class MpAccountProducer {
-
-    @Resource
-    private RedisMQTemplate redisMQTemplate;
-
-    /**
-     * 发送 {@link MpAccountRefreshMessage} 消息
-     */
-    public void sendAccountRefreshMessage() {
-        MpAccountRefreshMessage message = new MpAccountRefreshMessage();
-        redisMQTemplate.send(message);
-    }
-
-}
 package cn.iocoder.yudao.module.mp.service.account;
+import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.ObjUtil;
 import cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil;
 import cn.iocoder.yudao.framework.common.pojo.PageResult;
-import cn.iocoder.yudao.framework.common.util.collection.CollectionUtils;
 import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
 import cn.iocoder.yudao.module.mp.controller.admin.account.vo.MpAccountCreateReqVO;
 import cn.iocoder.yudao.module.mp.controller.admin.account.vo.MpAccountPageReqVO;
@@ -13,7 +13,6 @@ import cn.iocoder.yudao.module.mp.dal.dataobject.account.MpAccountDO;
 import cn.iocoder.yudao.module.mp.dal.mysql.account.MpAccountMapper;
 import cn.iocoder.yudao.module.mp.enums.ErrorCodeConstants;
 import cn.iocoder.yudao.module.mp.framework.mp.core.MpServiceFactory;
-import cn.iocoder.yudao.module.mp.mq.producer.MpAccountProducer;
 import com.google.common.annotations.VisibleForTesting;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -21,15 +20,20 @@ import me.chanjar.weixin.common.error.WxErrorException;
 import me.chanjar.weixin.mp.api.WxMpService;
 import me.chanjar.weixin.mp.bean.result.WxMpQrCodeTicket;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.validation.annotation.Validated;
 import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
+import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertMap;
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.getMaxValue;
 import static cn.iocoder.yudao.module.system.enums.ErrorCodeConstants.USER_USERNAME_EXISTS;
 /**
@@ -58,9 +62,6 @@ public class MpAccountServiceImpl implements MpAccountService {
     @Lazy // 延迟加载，解决循环依赖的问题
     private MpServiceFactory mpServiceFactory;
-    @Resource
-    private MpAccountProducer mpAccountProducer;
-
     @Override
     @PostConstruct
     public void initLocalCache() {
@@ -72,7 +73,30 @@ public class MpAccountServiceImpl implements MpAccountService {
             // 第二步：构建缓存。创建或更新支付 Client
             mpServiceFactory.init(accounts);
-            accountCache = CollectionUtils.convertMap(accounts, MpAccountDO::getAppId);
+            accountCache = convertMap(accounts, MpAccountDO::getAppId);
+        });
+    }
+
+    /**
+     * 通过定时任务轮询，刷新缓存
+     *
+     * 目的：多节点部署时，通过轮询”通知“所有节点，进行刷新
+     */
+    @Scheduled(initialDelay = 60, fixedRate = 60, timeUnit = TimeUnit.SECONDS)
+    public void refreshLocalCache() {
+        // 注意：忽略自动多租户，因为要全局初始化缓存
+        TenantUtils.executeIgnore(() -> {
+            // 情况一：如果缓存里没有数据，则直接刷新缓存
+            if (CollUtil.isEmpty(accountCache)) {
+                initLocalCache();
+                return;
+            }
+
+            // 情况二，如果缓存里数据，则通过 updateTime 判断是否有数据变更，有变更则刷新缓存
+            LocalDateTime maxTime = getMaxValue(accountCache.values(), MpAccountDO::getUpdateTime);
+            if (mpAccountMapper.selectCountByUpdateTimeGt(maxTime) > 0) {
+                initLocalCache();
+            }
         });
     }
@@ -85,8 +109,8 @@ public class MpAccountServiceImpl implements MpAccountService {
         MpAccountDO account = MpAccountConvert.INSTANCE.convert(createReqVO);
         mpAccountMapper.insert(account);
-        // 发送刷新消息
-        mpAccountProducer.sendAccountRefreshMessage();
+        // 刷新缓存
+        initLocalCache();
         return account.getId();
     }
@@ -101,8 +125,8 @@ public class MpAccountServiceImpl implements MpAccountService {
         MpAccountDO updateObj = MpAccountConvert.INSTANCE.convert(updateReqVO);
         mpAccountMapper.updateById(updateObj);
-        // 发送刷新消息
-        mpAccountProducer.sendAccountRefreshMessage();
+        // 刷新缓存
+        initLocalCache();
     }
     @Override
@@ -112,8 +136,8 @@ public class MpAccountServiceImpl implements MpAccountService {
         // 删除
         mpAccountMapper.deleteById(id);
-        // 发送刷新消息
-        mpAccountProducer.sendAccountRefreshMessage();
+        // 刷新缓存
+        initLocalCache();
     }
     private MpAccountDO validateAccountExists(Long id) {
@@ -6,8 +6,6 @@ import lombok.Data;
 /**
  * 支付单信息 Response DTO
  *
- * TODO 芋艿：还没定好字段
- *
  * @author 芋道源码
  */
 @Data
@@ -8,8 +8,6 @@ import java.time.LocalDateTime;
 /**
  * 退款单信息 Response DTO
  *
- * TODO 芋艿：还没定好字段
- *
  * @author 芋道源码
  */
 @Data
@@ -61,12 +61,6 @@
             <artifactId>yudao-spring-boot-starter-job</artifactId>
         </dependency>
-        <!-- 消息队列相关 -->
-        <dependency>
-            <groupId>cn.iocoder.boot</groupId>
-            <artifactId>yudao-spring-boot-starter-mq</artifactId>
-        </dependency>
-
         <!-- Test 测试相关 -->
         <dependency>
             <groupId>cn.iocoder.boot</groupId>
@@ -15,7 +15,6 @@ import cn.iocoder.yudao.module.pay.controller.admin.channel.vo.PayChannelUpdateR
 import cn.iocoder.yudao.module.pay.convert.channel.PayChannelConvert;
 import cn.iocoder.yudao.module.pay.dal.dataobject.channel.PayChannelDO;
 import cn.iocoder.yudao.module.pay.dal.mysql.channel.PayChannelMapper;
-import cn.iocoder.yudao.module.pay.enums.ErrorCodeConstants;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
@@ -112,7 +111,7 @@ public class PayChannelServiceImpl implements PayChannelService {
         channelMapper.insert(channel);
         // 刷新缓存
-        refreshLocalCache();
+        initLocalCache();
         return channel.getId();
     }
@@ -127,7 +126,7 @@ public class PayChannelServiceImpl implements PayChannelService {
         channelMapper.updateById(channel);
         // 刷新缓存
-        refreshLocalCache();
+        initLocalCache();
     }
     /**
@@ -160,7 +159,7 @@ public class PayChannelServiceImpl implements PayChannelService {
         channelMapper.deleteById(id);
         // 刷新缓存
-        refreshLocalCache();
+        initLocalCache();
     }
     private PayChannelDO validateChannelExists(Long id) {
@@ -167,7 +167,8 @@ yudao:
       - ${spring.boot.admin.context-path}/** # 不处理 Spring Boot Admin 的请求
       - ${management.endpoints.web.base-path}/** # 不处理 Actuator 的请求
   pay:
-    callback-url: http://yunai.natapp1.cc/admin-api/pay/notify/callback
+    order-notify-url: http://yunai.natapp1.cc/admin-api/pay/notify/order # 支付渠道的【支付】回调地址
+    refund-notify-url: http://yunai.natapp1.cc/admin-api/pay/notify/refund # 支付渠道的【退款】回调地址
   demo: true # 开启演示模式
 justauth: