art: 设备管理-三保-并行任务执行， 并行网关多任务处理

zhangherong

2025-05-27 e1856ede152bbe7ab32208716b87c7d289080221

 lxzn-module-flowable/src/main/java/org/jeecg/modules/flowable/service/impl/FlowDefinitionServiceImpl.java

@@ -1,6 +1,8 @@
package org.jeecg.modules.flowable.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
@@ -44,7 +46,6 @@

/**
 * 流程定义
 *
 */
@Service
public class FlowDefinitionServiceImpl extends FlowServiceFactory implements IFlowDefinitionService {
@@ -69,8 +70,8 @@
    /**
     * 流程定义列表
     *
     * @param pageNum  当前页码
     * @param pageSize 每页条数
     * @param pageNum        当前页码
     * @param pageSize       每页条数
     * @param flowProcDefDto
     * @return 流程定义分页列表数据
     */
@@ -83,19 +84,19 @@
                //.processDefinitionId("cs:5:15e953ed-4d09-11ec-85b8-e884a5deddfc")
                .orderByProcessDefinitionKey().asc().orderByProcessDefinitionVersion().desc();
        /*=====参数=====*/
        if (StrUtil.isNotBlank(flowProcDefDto.getName())){
            processDefinitionQuery.processDefinitionNameLike("%"+flowProcDefDto.getName()+"%");
        if (StrUtil.isNotBlank(flowProcDefDto.getName())) {
            processDefinitionQuery.processDefinitionNameLike("%" + flowProcDefDto.getName() + "%");
        }
        if (StrUtil.isNotBlank(flowProcDefDto.getCategory())){
        if (StrUtil.isNotBlank(flowProcDefDto.getCategory())) {
            processDefinitionQuery.processDefinitionCategory(flowProcDefDto.getCategory());
        }
        if (flowProcDefDto.getSuspensionState() == 1){
        if (flowProcDefDto.getSuspensionState() == 1) {
            processDefinitionQuery.active();
        }
        if (StrUtil.isNotBlank(flowProcDefDto.getKey())){
        if (StrUtil.isNotBlank(flowProcDefDto.getKey())) {
            processDefinitionQuery.processDefinitionKey(flowProcDefDto.getKey());
        }
        if (flowProcDefDto.getIsLastVersion() == 1){
        if (flowProcDefDto.getIsLastVersion() == 1) {
            processDefinitionQuery.latestVersion();
        }
        /*============*/
@@ -149,7 +150,7 @@
    @Override
    public Result readXmlByDataId(String dataId) throws IOException {
        LambdaQueryWrapper<FlowMyBusiness> flowMyBusinessLambdaQueryWrapper = new LambdaQueryWrapper<>();
        flowMyBusinessLambdaQueryWrapper.eq(FlowMyBusiness::getDataId,dataId)
        flowMyBusinessLambdaQueryWrapper.eq(FlowMyBusiness::getDataId, dataId)
        ;
        //如果保存数据前未调用必调的FlowCommonService.initActBusiness方法，就会有问题
        FlowMyBusiness business = flowMyBusinessService.getOne(flowMyBusinessLambdaQueryWrapper);
@@ -201,6 +202,7 @@
        InputStream in = processDiagramGenerator.generateDiagram(bpmnModel, "png", activityIds, flows, engConf.getActivityFontName(), engConf.getLabelFontName(), engConf.getAnnotationFontName(), engConf.getClassLoader(), 1.0, true);
        return in;
    }

    /**
     * 读取xml
     *
@@ -232,7 +234,7 @@
     * 根据流程定义ID启动流程实例
     *
     * @param procDefKey 流程定义Id
     * @param variables 流程变量
     * @param variables  流程变量
     * @return
     */
    @Override
@@ -241,9 +243,10 @@
        ProcessDefinition processDefinition = repositoryService.createProcessDefinitionQuery()
                .processDefinitionKey(procDefKey)
                .latestVersion().singleResult();
        Result result = startProcessInstanceById(processDefinition.getId(),variables);
        Result result = startProcessInstanceById(processDefinition.getId(), variables);
        return result;
    }

    /**
     * 根据流程定义ID启动流程实例
     *
@@ -254,28 +257,28 @@
    @Override
    @Transactional
    public Result<?> startProcessInstanceById(String procDefId, Map<String, Object> variables) {
            ProcessDefinition processDefinition = repositoryService.createProcessDefinitionQuery()
                    .processDefinitionId(procDefId)
                    .singleResult();
            if (Objects.nonNull(processDefinition) && processDefinition.isSuspended()) {
                return Result.error("流程已被挂起,请先激活流程");
            }
        ProcessDefinition processDefinition = repositoryService.createProcessDefinitionQuery()
                .processDefinitionId(procDefId)
                .singleResult();
        if (Objects.nonNull(processDefinition) && processDefinition.isSuspended()) {
            return Result.error("流程已被挂起,请先激活流程");
        }
//           variables.put("skip", true);
//           variables.put(ProcessConstants.FLOWABLE_SKIP_EXPRESSION_ENABLED, true);
            // 设置流程发起人Id到流程中
            SysUser sysUser = iFlowThirdService.getLoginUser();
            identityService.setAuthenticatedUserId(sysUser.getUsername());
            variables.put(ProcessConstants.PROCESS_INITIATOR, sysUser.getUsername());
            ProcessInstance processInstance = runtimeService.startProcessInstanceById(procDefId, variables);
            // 给第一步申请人节点设置任务执行人和意见
            Task task = taskService.createTaskQuery().processInstanceId(processInstance.getProcessInstanceId()).active().singleResult();
            if (Objects.nonNull(task)) {
                taskService.addComment(task.getId(), processInstance.getProcessInstanceId(), FlowComment.NORMAL.getType(), variables.get("organization").toString());
                taskService.setAssignee(task.getId(), sysUser.getUsername());
                task.setDescription(variables.get("organization").toString());
                //taskService.complete(task.getId(), variables);
            }
            //设置所有申请人
        // 设置流程发起人Id到流程中
        SysUser sysUser = iFlowThirdService.getLoginUser();
        identityService.setAuthenticatedUserId(sysUser.getUsername());
        variables.put(ProcessConstants.PROCESS_INITIATOR, sysUser.getUsername());
        ProcessInstance processInstance = runtimeService.startProcessInstanceById(procDefId, variables);
        // 给第一步申请人节点设置任务执行人和意见
        Task task = taskService.createTaskQuery().processInstanceId(processInstance.getProcessInstanceId()).active().singleResult();
        if (Objects.nonNull(task)) {
            taskService.addComment(task.getId(), processInstance.getProcessInstanceId(), FlowComment.NORMAL.getType(), variables.get("organization").toString());
            taskService.setAssignee(task.getId(), sysUser.getUsername());
            task.setDescription(variables.get("organization").toString());
            //taskService.complete(task.getId(), variables);
        }
        //设置所有申请人

        /*======================todo 启动之后  回调以及关键数据保存======================*/
        //业务数据id
@@ -283,77 +286,146 @@
        //如果保存数据前未调用必调的FlowCommonService.initActBusiness方法，就会有问题
        FlowMyBusiness business = flowMyBusinessService.getByDataId(dataId);
        //设置数据
        FlowNextDto nextFlowNode = flowTaskService.getNextFlowNode(task.getId(), variables);
        List<FlowNextDto> nextFlowNodeList = flowTaskService.getNextFlowNode(task.getId(), variables);
        taskService.complete(task.getId(), variables);
        //下一个实例节点  多实例会是一个list,随意取一个即可  数组中定义Key是一致的
        //Task task2 = taskService.createTaskQuery().processInstanceId(processInstance.getProcessInstanceId()).active().singleResult();
        List<Task> task2List = taskService.createTaskQuery().processInstanceId(processInstance.getProcessInstanceId()).active().list();
        Task task2 = null;
        if(task2List.size()>0) task2 = task2List.get(0);
        String doneUsers = business.getDoneUsers();
        // 处理过流程的人
        JSONArray doneUserList = new JSONArray();
        if (StrUtil.isNotBlank(doneUsers)){
        if (StrUtil.isNotBlank(doneUsers)) {
            doneUserList = JSON.parseArray(doneUsers);
        }
        if (!doneUserList.contains(sysUser.getUsername())){
        if (!doneUserList.contains(sysUser.getUsername())) {
            doneUserList.add(sysUser.getUsername());
        }

        if (nextFlowNode!=null){
            //**有下一个节点
            UserTask nextTask = nextFlowNode.getUserTask();
            //能够处理下个节点的候选人
            List<SysUser> nextFlowNodeUserList = nextFlowNode.getUserList();

            List<String> collect_username = nextFlowNodeUserList.stream().map(SysUser::getUsername).collect(Collectors.toList());
            //spring容器类名
            String serviceImplName = business.getServiceImplName();
            FlowCallBackServiceI flowCallBackService = (FlowCallBackServiceI) SpringContextUtils.getBean(serviceImplName);
            List<String> beforeParamsCandidateUsernames = flowCallBackService.flowCandidateUsernamesOfTask(task2.getTaskDefinitionKey(), variables);
            business.setProcessDefinitionId(procDefId)
                    .setProcessInstanceId(processInstance.getProcessInstanceId())
////                    .setActStatus(ActStatus.start)
                    .setProposer(sysUser.getUsername())
                    .setTaskId(task2.getId())
                    .setTaskName(nextTask.getName())
                    .setTaskNameId(nextTask.getId())
                    .setPriority(nextTask.getPriority())
                    .setDoneUsers(doneUserList.toJSONString());
            if (CollUtil.isNotEmpty(beforeParamsCandidateUsernames)){
                // 删除后重写
                for (Task task2One : task2List) {
                    for (String oldUser : collect_username) {
                        taskService.deleteCandidateUser(task2One.getId(),oldUser);
                    }
                }
                // 业务层有指定候选人，覆盖
                for (Task task2One : task2List) {
                    for (String newUser : beforeParamsCandidateUsernames) {
                        taskService.addCandidateUser(task2One.getId(),newUser);
                    }
                }
                business.setTodoUsers(JSON.toJSONString(beforeParamsCandidateUsernames));
            }else {
                // 业务层没有指定候选人，直接覆盖
                business.setTodoUsers(JSON.toJSONString(collect_username));
            }
        } else {
        //    **没有下一个节点，流程已经结束了
        if (CollectionUtil.isEmpty(nextFlowNodeList)) {
            //    **没有下一个节点，流程已经结束了
            business.setProcessDefinitionId(procDefId)
                    .setProcessInstanceId(processInstance.getProcessInstanceId())
//                    .setActStatus(ActStatus.pass)
                    .setProposer(sysUser.getUsername())
                    .setDoneUsers(doneUserList.toJSONString())
            ;
            flowMyBusinessService.updateById(business);
        } else if (nextFlowNodeList.size() == 1) {
            //下个节点只有一个
            FlowNextDto nextFlowNode = nextFlowNodeList.get(0);
            UserTask nextTask = nextFlowNode.getUserTask();
            Task task2 = null;
            Optional<Task> first = task2List.stream().filter(t -> t.getTaskDefinitionKey().equals(nextTask.getId())).findFirst();
            if (first.isPresent()) {
                task2 = first.get();
            }
            if (task2 != null) {
                //能够处理下个节点的候选人
                List<SysUser> nextFlowNodeUserList = nextFlowNode.getUserList();

                List<String> collect_username = nextFlowNodeUserList.stream().map(SysUser::getUsername).collect(Collectors.toList());
                //spring容器类名
                String serviceImplName = business.getServiceImplName();
                FlowCallBackServiceI flowCallBackService = (FlowCallBackServiceI) SpringContextUtils.getBean(serviceImplName);
                List<String> beforeParamsCandidateUsernames = flowCallBackService.flowCandidateUsernamesOfTask(task2.getTaskDefinitionId(), variables);
                business.setProcessDefinitionId(procDefId)
                        .setProcessInstanceId(processInstance.getProcessInstanceId())
////                    .setActStatus(ActStatus.start)
                        .setProposer(sysUser.getUsername())
                        .setTaskId(task2.getId())
                        .setTaskName(nextTask.getName())
                        .setTaskNameId(nextTask.getId())
                        .setPriority(nextTask.getPriority())
                        .setDoneUsers(doneUserList.toJSONString());
                if (CollUtil.isNotEmpty(beforeParamsCandidateUsernames)) {
                    // 删除后重写
                    for (String oldUser : collect_username) {
                        taskService.deleteCandidateUser(task2.getId(), oldUser);
                    }
                    // 业务层有指定候选人，覆盖
                    for (String newUser : beforeParamsCandidateUsernames) {
                        taskService.addCandidateUser(task2.getId(), newUser);
                    }
                    business.setTodoUsers(JSON.toJSONString(beforeParamsCandidateUsernames));
                } else {
                    // 业务层没有指定候选人，直接覆盖
                    business.setTodoUsers(JSON.toJSONString(collect_username));
                }
                flowMyBusinessService.updateById(business);
            }
        } else {
            for (FlowNextDto nextFlowNode : nextFlowNodeList) {
                //**有下一个节点
                UserTask nextTask = nextFlowNode.getUserTask();
                Task task2 = null;
                Optional<Task> first = task2List.stream().filter(t -> t.getTaskDefinitionKey().equals(nextTask.getId())).findFirst();
                if (first.isPresent()) {
                    task2 = first.get();
                }
                if (task2 != null) {
                    //新的业务流程节点
                    FlowMyBusiness newBusiness =  BeanUtil.copyProperties(business, FlowMyBusiness.class, "id");
                    //能够处理下个节点的候选人
                    List<SysUser> nextFlowNodeUserList = nextFlowNode.getUserList();

                    List<String> collect_username = nextFlowNodeUserList.stream().map(SysUser::getUsername).collect(Collectors.toList());
                    //spring容器类名
                    String serviceImplName = newBusiness.getServiceImplName();
                    FlowCallBackServiceI flowCallBackService = (FlowCallBackServiceI) SpringContextUtils.getBean(serviceImplName);
                    List<String> beforeParamsCandidateUsernames = flowCallBackService.flowCandidateUsernamesOfTask(task2.getTaskDefinitionId(), variables);
                    newBusiness.setProcessDefinitionId(procDefId)
                            .setProcessInstanceId(processInstance.getProcessInstanceId())
////                    .setActStatus(ActStatus.start)
                            .setProposer(sysUser.getUsername())
                            .setTaskId(task2.getId())
                            .setTaskName(nextTask.getName())
                            .setTaskNameId(nextTask.getId())
                            .setPriority(nextTask.getPriority())
                            .setDoneUsers(doneUserList.toJSONString());
                    if (CollUtil.isNotEmpty(beforeParamsCandidateUsernames)) {
                        // 删除后重写
                        for (Task task2One : task2List) {
                            for (String oldUser : beforeParamsCandidateUsernames) {
                                taskService.deleteCandidateUser(task2One.getId(), oldUser);
                            }
                        }
                        // 业务层有指定候选人，覆盖
                        for (Task task2One : task2List) {
                            for (String newUser : beforeParamsCandidateUsernames) {
                                taskService.addCandidateUser(task2One.getId(), newUser);
                            }
                        }
                        newBusiness.setTodoUsers(JSON.toJSONString(beforeParamsCandidateUsernames));
                    } else {
                        // 业务层没有指定候选人，直接覆盖
                        newBusiness.setTodoUsers(JSON.toJSONString(collect_username));
                        // 删除后重写
                        for (Task task2One : task2List) {
                            for (String oldUser : collect_username) {
                                taskService.deleteCandidateUser(task2One.getId(), oldUser);
                            }
                        }
                        // 业务层有指定候选人，覆盖
                        for (Task task2One : task2List) {
                            for (String newUser : collect_username) {
                                taskService.addCandidateUser(task2One.getId(), newUser);
                            }
                        }
                    }
                    flowMyBusinessService.save(newBusiness);
                }
            }
            //删除原有的业务数据
            flowMyBusinessService.removeById(business.getId());
        }
        flowMyBusinessService.updateById(business);
        //spring容器类名
        String serviceImplName = business.getServiceImplName();
        FlowCallBackServiceI flowCallBackService = (FlowCallBackServiceI) SpringContextUtils.getBean(serviceImplName);
        // 流程处理完后，进行回调业务层
        business.setValues(variables);
        if (flowCallBackService!=null)flowCallBackService.afterFlowHandle(business);
        if (flowCallBackService != null) {
            flowCallBackService.afterFlowHandle(business);
        }
        return Result.OK("流程启动成功");
    }

@@ -363,13 +435,13 @@
        flowMyBusinessLambdaQueryWrapper.eq(FlowMyBusiness::getDataId, dataId)
        ;
        FlowMyBusiness business = flowMyBusinessService.getOne(flowMyBusinessLambdaQueryWrapper);
        if (business==null){
            return Result.error("未找到dataId："+dataId);
        if (business == null) {
            return Result.error("未找到dataId：" + dataId);
        }
        if (StrUtil.isNotBlank(business.getProcessDefinitionId())){
            return this.startProcessInstanceById(business.getProcessDefinitionId(),variables);
        if (StrUtil.isNotBlank(business.getProcessDefinitionId())) {
            return this.startProcessInstanceById(business.getProcessDefinitionId(), variables);
        }
        return this.startProcessInstanceByKey(business.getProcessDefinitionKey(),variables);
        return this.startProcessInstanceByKey(business.getProcessDefinitionKey(), variables);
    }