activiti部署流程deployment实现源码实现过程

以你之姓@ 2022-05-08 03:38 1051阅读 0赞

以下是流程部署的实现过程，看了一天源码了解了activiti源码是如何实现的，包括数据数据，了解activiti的架构

//根据bpmn文件部署流程
    Deployment deployment = repositoryService.createDeployment().addClasspathResource("diagrams/demo2.bpmn").deploy();

调用顺序

//RepositoryServiceImpl类的方法
    public Deployment deploy(DeploymentBuilderImpl deploymentBuilder) {
            return (Deployment)this.commandExecutor.execute(new DeployCmd(deploymentBuilder));
        }
    
    
    //CommandExecutorImpl类的方法
    public <T> T execute(Command<T> command) {
            return this.execute(this.defaultConfig, command);
        }
    
    public <T> T execute(CommandConfig config, Command<T> command) {
            return this.first.execute(config, command);
        }

这里的命令执行器CommandExecutorImpl属性first是接口CommandInterceptor，执行顺序是LogInterceptor、SpringTransactionInterceptor、XX、CommandInvoker。

LogInterceptor只是输出日志

SpringTransactionInterceptor是控制事务的，因为部署的时候会往表ACT\_RE\_PROCDEF、ACT\_RE\_DEPLOYMENT、ACT\_GE\_BYTEARRAY同时插入数据，保证事务完整性。

![70][]

![70 1][]

CommandContextInterceptor

CommandInvoker是执行传入的command的excute方法，这里传入的是DeployCmd，所以去看DeployCmd的excute方法

public Deployment execute(CommandContext commandContext) {
        DeploymentEntity deployment = deploymentBuilder.getDeployment();
    
        deployment.setDeploymentTime(Context.getProcessEngineConfiguration().getClock().getCurrentTime());
    
        if ( deploymentBuilder.isDuplicateFilterEnabled() ) {
          DeploymentEntity existingDeployment = Context
            .getCommandContext()
            .getDeploymentEntityManager()
            .findLatestDeploymentByName(deployment.getName());
          
          if ( (existingDeployment!=null)
               && !deploymentsDiffer(deployment, existingDeployment)) {
            return existingDeployment;
          }
        }
    
        deployment.setNew(true);
        
        // Save the data
        Context
          .getCommandContext()
          .getDeploymentEntityManager()
          .insertDeployment(deployment);
        
        if(Context.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
    	    Context.getProcessEngineConfiguration().getEventDispatcher().dispatchEvent(
    	    		ActivitiEventBuilder.createEntityEvent(ActivitiEventType.ENTITY_CREATED, deployment));
        }
        
        // Deployment settings
        Map<String, Object> deploymentSettings = new HashMap<String, Object>();
        deploymentSettings.put(DeploymentSettings.IS_BPMN20_XSD_VALIDATION_ENABLED, deploymentBuilder.isBpmn20XsdValidationEnabled());
        deploymentSettings.put(DeploymentSettings.IS_PROCESS_VALIDATION_ENABLED, deploymentBuilder.isProcessValidationEnabled());
        
        // Actually deploy
        Context
          .getProcessEngineConfiguration()
          .getDeploymentManager()
          .deploy(deployment, deploymentSettings);
        
        if (deploymentBuilder.getProcessDefinitionsActivationDate() != null) {
          scheduleProcessDefinitionActivation(commandContext, deployment);
        }
        
        if(Context.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
    	    Context.getProcessEngineConfiguration().getEventDispatcher().dispatchEvent(
    	    		ActivitiEventBuilder.createEntityEvent(ActivitiEventType.ENTITY_INITIALIZED, deployment));
        }
        
        return deployment;
      }

这里保存DeploymentEntity
    Context.getCommandContext().getDeploymentEntityManager().insertDeployment(deployment);

![70 2][]

再来看DbSqlSession里的insert方法，这里会把实体存入Map insertedObjects里，之后会用到

public void insert(PersistentObject persistentObject) {
            if (persistentObject.getId() == null) {
                String id = this.dbSqlSessionFactory.getIdGenerator().getNextId();
                persistentObject.setId(id);
            }
    
            Class<? extends PersistentObject> clazz = persistentObject.getClass();
            if (!this.insertedObjects.containsKey(clazz)) {
                this.insertedObjects.put(clazz, new ArrayList());
            }
    
            ((List)this.insertedObjects.get(clazz)).add(persistentObject);
            this.cachePut(persistentObject, false);
        }

再回到DeployCmd接着往下走，实际部署。

Context.getProcessEngineConfiguration().getDeploymentManager().deploy(deployment, deploymentSettings);

执行的是BpmnDeployer里的deploy方法

public void deploy(DeploymentEntity deployment, Map<String, Object> deploymentSettings) {
        LOG.debug("Processing deployment {}", deployment.getName());
        
        List<ProcessDefinitionEntity> processDefinitions = new ArrayList<ProcessDefinitionEntity>();
        Map<String, ResourceEntity> resources = deployment.getResources();
    
        for (String resourceName : resources.keySet()) {
    
          LOG.info("Processing resource {}", resourceName);
          if (isBpmnResource(resourceName)) {
            ResourceEntity resource = resources.get(resourceName);
            byte[] bytes = resource.getBytes();
            ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
            
            BpmnParse bpmnParse = bpmnParser
              .createParse()
              .sourceInputStream(inputStream)
              .deployment(deployment)
              .name(resourceName);
            
            if (deploymentSettings != null) {
            	
            	// Schema validation if needed
            	if (deploymentSettings.containsKey(DeploymentSettings.IS_BPMN20_XSD_VALIDATION_ENABLED)) {
            		bpmnParse.setValidateSchema((Boolean) deploymentSettings.get(DeploymentSettings.IS_BPMN20_XSD_VALIDATION_ENABLED));
            	}
            	
            	// Process validation if needed
            	if (deploymentSettings.containsKey(DeploymentSettings.IS_PROCESS_VALIDATION_ENABLED)) {
            		bpmnParse.setValidateProcess((Boolean) deploymentSettings.get(DeploymentSettings.IS_PROCESS_VALIDATION_ENABLED));
            	}
            	
            }
            
            bpmnParse.execute();
            
            for (ProcessDefinitionEntity processDefinition: bpmnParse.getProcessDefinitions()) {
              processDefinition.setResourceName(resourceName);
              
              if (deployment.getTenantId() != null) {
              	processDefinition.setTenantId(deployment.getTenantId()); // process definition inherits the tenant id
              }
              
              String diagramResourceName = getDiagramResourceForProcess(resourceName, processDefinition.getKey(), resources);
                       
              // Only generate the resource when deployment is new to prevent modification of deployment resources 
              // after the process-definition is actually deployed. Also to prevent resource-generation failure every
              // time the process definition is added to the deployment-cache when diagram-generation has failed the first time.
              if(deployment.isNew()) {
                if (Context.getProcessEngineConfiguration().isCreateDiagramOnDeploy() &&
                      diagramResourceName==null && processDefinition.isGraphicalNotationDefined()) {
                  try {
                      byte[] diagramBytes = IoUtil.readInputStream(ProcessDiagramGenerator.generatePngDiagram(bpmnParse.getBpmnModel()), null);
                      diagramResourceName = getProcessImageResourceName(resourceName, processDefinition.getKey(), "png");
                      createResource(diagramResourceName, diagramBytes, deployment);
                  } catch (Throwable t) { // if anything goes wrong, we don't store the image (the process will still be executable).
                    LOG.warn("Error while generating process diagram, image will not be stored in repository", t);
                  }
                } 
              }
              
              processDefinition.setDiagramResourceName(diagramResourceName);
              processDefinitions.add(processDefinition);
            }
          }
        }
        
        // check if there are process definitions with the same process key to prevent database unique index violation
        List<String> keyList = new ArrayList<String>();
        for (ProcessDefinitionEntity processDefinition : processDefinitions) {
          if (keyList.contains(processDefinition.getKey())) {
            throw new ActivitiException("The deployment contains process definitions with the same key (process id atrribute), this is not allowed");
          }
          keyList.add(processDefinition.getKey());
        }
        
        CommandContext commandContext = Context.getCommandContext();
        ProcessDefinitionEntityManager processDefinitionManager = commandContext.getProcessDefinitionEntityManager();
        DbSqlSession dbSqlSession = commandContext.getSession(DbSqlSession.class);
        for (ProcessDefinitionEntity processDefinition : processDefinitions) {
          List<TimerEntity> timers = new ArrayList<TimerEntity>();
          if (deployment.isNew()) {
            int processDefinitionVersion;
    
            ProcessDefinitionEntity latestProcessDefinition = null;
            if (processDefinition.getTenantId() != null && !ProcessEngineConfiguration.NO_TENANT_ID.equals(processDefinition.getTenantId())) {
            	latestProcessDefinition = processDefinitionManager
            			.findLatestProcessDefinitionByKeyAndTenantId(processDefinition.getKey(), processDefinition.getTenantId());
            } else {
            	latestProcessDefinition = processDefinitionManager
            			.findLatestProcessDefinitionByKey(processDefinition.getKey());
            }
            		
            if (latestProcessDefinition != null) {
              processDefinitionVersion = latestProcessDefinition.getVersion() + 1;
            } else {
              processDefinitionVersion = 1;
            }
    
            processDefinition.setVersion(processDefinitionVersion);
            processDefinition.setDeploymentId(deployment.getId());
    
            String nextId = idGenerator.getNextId();
            String processDefinitionId = processDefinition.getKey() 
              + ":" + processDefinition.getVersion()
              + ":" + nextId; // ACT-505
                       
            // ACT-115: maximum id length is 64 charcaters
            if (processDefinitionId.length() > 64) {          
              processDefinitionId = nextId; 
            }
            processDefinition.setId(processDefinitionId);
            
            if(commandContext.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
            	commandContext.getProcessEngineConfiguration().getEventDispatcher().dispatchEvent(
            			ActivitiEventBuilder.createEntityEvent(ActivitiEventType.ENTITY_CREATED, processDefinition));
            }
    
            removeObsoleteTimers(processDefinition);
            addTimerDeclarations(processDefinition, timers);
            
            removeObsoleteMessageEventSubscriptions(processDefinition, latestProcessDefinition);
            addMessageEventSubscriptions(processDefinition);
            
            removeObsoleteSignalEventSubScription(processDefinition, latestProcessDefinition);
            addSignalEventSubscriptions(processDefinition);
    
            dbSqlSession.insert(processDefinition);
            addAuthorizations(processDefinition);
    
            if(commandContext.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
            	commandContext.getProcessEngineConfiguration().getEventDispatcher().dispatchEvent(
            			ActivitiEventBuilder.createEntityEvent(ActivitiEventType.ENTITY_INITIALIZED, processDefinition));
            }
    
            scheduleTimers(timers);
    
          } else {
            String deploymentId = deployment.getId();
            processDefinition.setDeploymentId(deploymentId);
            
            ProcessDefinitionEntity persistedProcessDefinition = null; 
            if (processDefinition.getTenantId() == null || ProcessEngineConfiguration.NO_TENANT_ID.equals(processDefinition.getTenantId())) {
            	persistedProcessDefinition = processDefinitionManager.findProcessDefinitionByDeploymentAndKey(deploymentId, processDefinition.getKey());
            } else {
            	persistedProcessDefinition = processDefinitionManager.findProcessDefinitionByDeploymentAndKeyAndTenantId(deploymentId, processDefinition.getKey(), processDefinition.getTenantId());
            }
            
            if (persistedProcessDefinition != null) {
            	processDefinition.setId(persistedProcessDefinition.getId());
            	processDefinition.setVersion(persistedProcessDefinition.getVersion());
            	processDefinition.setSuspensionState(persistedProcessDefinition.getSuspensionState());
            }
          }
    
          // Add to cache
          Context
            .getProcessEngineConfiguration()
            .getDeploymentManager()
            .getProcessDefinitionCache()
            .add(processDefinition.getId(), processDefinition);
          
          // Add to deployment for further usage
          deployment.addDeployedArtifact(processDefinition);
        }
      }

这里主要看bpmnParse.execute();
    这里相当于把xml的内容封装成对象bpmnModel
    bpmnModel = converter.convertToBpmnModel(streamSource, validateSchema, enableSafeBpmnXml, encoding);
    然后看方法transformProcessDefinitions()，xml里的每一个<process>标签就是一个流程定义。

![70 3][]

protected void transformProcessDefinitions() {
        sequenceFlows = new HashMap<String, TransitionImpl>();
        for (Process process : bpmnModel.getProcesses()) {
          if (process.isExecutable()) {
            bpmnParserHandlers.parseElement(this, process);
          }
        }
    
        if (processDefinitions.size() > 0) {
          processDI();
        }
      }

再来看bpmnParserHandlers.parseElement(this, process);

![70 4][]

ProcessParseHandler 这样流程定义就存入bpmnParse的ProcessDefinitions集合。

protected void executeParse(BpmnParse bpmnParse, Process process) {
        if (process.isExecutable() == false) {
          LOGGER.info("Ignoring non-executable process with id='" + process.getId() + "'. Set the attribute isExecutable=\"true\" to deploy this process.");
        } else {
          bpmnParse.getProcessDefinitions().add(transformProcess(bpmnParse, process));
        }
      }

ProcessHistoryParseHandler，监听流程结束的时候更新流程实例。

protected void executeParse(BpmnParse bpmnParse, Process element) {
        bpmnParse.getCurrentProcessDefinition().addExecutionListener(org.activiti.engine.impl.pvm.PvmEvent.EVENTNAME_END, PROCESS_INSTANCE_END_HANDLER);
      }

再回到transformProcessDefinitions方法往下走看processDI方法，这里是根据xml回话png图片

public void processDI() {
        if (bpmnModel.getLocationMap().size() > 0) {
    
          // Verify if all referenced elements exist
          for (String bpmnReference : bpmnModel.getLocationMap().keySet()) {
            if (bpmnModel.getFlowElement(bpmnReference) == null) {
            	// ACT-1625: don't warn when	artifacts are referenced from DI
            	if(bpmnModel.getArtifact(bpmnReference) == null) {
            		LOGGER.warn("Invalid reference in diagram interchange definition: could not find " + bpmnReference);
            	}
            } else if (! (bpmnModel.getFlowElement(bpmnReference) instanceof FlowNode)) {
              LOGGER.warn("Invalid reference in diagram interchange definition: " + bpmnReference + " does not reference a flow node");
            }
          }
          for (String bpmnReference : bpmnModel.getFlowLocationMap().keySet()) {
            if (bpmnModel.getFlowElement(bpmnReference) == null) {
              // ACT-1625: don't warn when	artifacts are referenced from DI
            	if(bpmnModel.getArtifact(bpmnReference) == null) {
            		LOGGER.warn("Invalid reference in diagram interchange definition: could not find " + bpmnReference);
            	}	
            } else if (! (bpmnModel.getFlowElement(bpmnReference) instanceof SequenceFlow)) {
              if (bpmnModel.getFlowLocationMap().get(bpmnReference).size() > 0) {
                LOGGER.warn("Invalid reference in diagram interchange definition: " + bpmnReference + " does not reference a sequence flow");
              } else {
                LOGGER.warn("Invalid reference in diagram interchange definition: " + bpmnReference + " does not reference a sequence flow");
              }
            }
          }
          
          for (Process process : bpmnModel.getProcesses()) {
            if (!process.isExecutable()) {
              continue;
            }
            
            // Parse diagram interchange information
            ProcessDefinitionEntity processDefinition = getProcessDefinition(process.getId());
            if (processDefinition != null) {
              processDefinition.setGraphicalNotationDefined(true);
              for (String shapeId : bpmnModel.getLocationMap().keySet()) {
                if (processDefinition.findActivity(shapeId) != null) {
                  createBPMNShape(shapeId, bpmnModel.getGraphicInfo(shapeId), processDefinition);
                }
              }
    
              for (String edgeId : bpmnModel.getFlowLocationMap().keySet()) {
                if (bpmnModel.getFlowElement(edgeId) != null) {
                  createBPMNEdge(edgeId, bpmnModel.getFlowLocationGraphicInfo(edgeId));
                }
              }
            }
          }
        }
      }

这里看bpmModel的locationMap，里面存的是每个任务节点的坐标。这样就可以取出locationMap的任务节点坐标存入

processDefinition.namedActivities里对应key的任务节点对应ActivityImpl坐标

![70 5][]

flowLocationMap里面存的是箭头的起点和终点坐标。从bpmnModel.process.flowElementList根据key取出对应的FlowElement。然后取出bmpnParse.sequenceFlows根据key取出对应的sequenceFlow。之后从flowLocationMap取出的坐标存入list然后存入sequenceFlow.waypoints。

![70 6][]

再回到BpmnDeploy.deploy方法往下走。绘制成png的流程图，拼接png的路径，生成ResourceEntity存入DbSqlSession。最后把生成的processDefinition添加入集合processDefinitions。之后循环processDefinitions，set基本参数，然后存入DbSqlSession。此时就deploy结束，这时的DbSqlSession的insertedObjects有两个ResourceEntity，一个DeploymentEntity,一个ProcessDefinitionEntity。

![70 7][]

结束deploy后回到CommandContextInterceptor 里的context.close方法，主要是执行方法flushSessions()方法

![70 8][]

最后是执行DbSqlSession里的flush方法

![70 9][]

之前在源码里根据mybatis的ProcessDefinition.xml根据插入语句的insertProcessDefinition搜索，没有搜索到，原来是”insertProcessDefinition“这个statement是拼出来的。之后再用mybatis插入这四条数据。

![70 10][]

四条数据执行完成后再提交事务，最后流程的部署就完成了。

[70]: /images/20220508/b43274dcab094daebba5c29093f15881.png
[70 1]: /images/20220508/beb8daa6e08643bf8cf95bdb66180c4b.png
[70 2]: /images/20220508/4de6af726e5f49e89e4188ca549a1ae3.png
[70 3]: /images/20220508/919c1dadf5c14da29d5c707863655ded.png
[70 4]: /images/20220508/7388505190b5405981354396de856858.png
[70 5]: /images/20220508/e004fd05c1fd4fac8cd6023b92af117d.png
[70 6]: /images/20220508/04c6d2311ee34dffbaf97f0f7e8ed69d.png
[70 7]: /images/20220508/40682199483b48fdb35167e3d3959695.png
[70 8]: /images/20220508/ea86eb3f71604b0c8a615895c72c451f.png
[70 9]: /images/20220508/7847bdc0f7c946709aca9500275bc8c6.png
[70 10]: /images/20220508/a771fbcbcddf429e9d57b5e2838d0763.png