[Java知识库]一、activiti如何加载流程文件

activiti5.0与springboot集成

添加依赖

    <dependency>
        <groupId>org.activiti</groupId>
        <artifactId>activiti-spring-boot-starter-basic</artifactId>
        <version>5.22.0</version>
    </dependency>

添加配置,以yml文件为例，数据源为mysql

activiti:
  #关闭定时任务JOB
  async-executor-activate: false
  check-process-definitions: false
  database-schema-update: true

可能出现的问题

• 报processes文件夹不存在，这个文件夹是放生成的bpmn文件的，可以在resources文件夹下新建processes或者指定bpmn的文件夹路径

process-definition-location-prefix: classpath*:/diagrams/

• 报找不到表，activiti总共有23张表，在启动的时候会校验，我们可以拿脚本自己去跑，或者将配置文件中database-schema-update改为true
  这样启动依旧会报错，我们需要在mysql的连接后面加上nullCatalogMeansCurrent=true，配置如下

   jdbc-url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8
&nullCatalogMeansCurrent=true
   username: root
   password: 123456
   driver-class-name: com.mysql.jdbc.Driver

activiti启动加载

加载activiti-spring-boot-starter-basic包下的spring.factories

    org.activiti.spring.boot.DataSourceProcessEngineAutoConfiguration,\
    org.activiti.spring.boot.EndpointAutoConfiguration,\
    org.activiti.spring.boot.RestApiAutoConfiguration,\
    org.activiti.spring.boot.JpaProcessEngineAutoConfiguration,\
    org.activiti.spring.boot.SecurityAutoConfiguration

DataSourceProcessEngineAutoConfiguration解析

• 加载SpringProcessEngineConfiguration

   //ConditionalOnMissingBean的意思是只加载一个bean
   @Bean
   @ConditionalOnMissingBean
   public SpringProcessEngineConfiguration springProcessEngineConfiguration(
           DataSource dataSource,
           PlatformTransactionManager transactionManager,
           SpringAsyncExecutor springAsyncExecutor) throws IOException {
     
     return this.baseSpringProcessEngineConfiguration(dataSource, transactionManager, springAsyncExecutor);
   }
   //初始化配置
 protected SpringProcessEngineConfiguration baseSpringProcessEngineConfiguration(DataSource dataSource, PlatformTransactionManager platformTransactionManager,
                                                                                 SpringAsyncExecutor springAsyncExecutor) throws IOException {

   List<Resource> procDefResources = this.discoverProcessDefinitionResources(
       this.resourceLoader, this.activitiProperties.getProcessDefinitionLocationPrefix(),
       this.activitiProperties.getProcessDefinitionLocationSuffixes(),
       this.activitiProperties.isCheckProcessDefinitions());

   SpringProcessEngineConfiguration conf = super.processEngineConfigurationBean(
       procDefResources.toArray(new Resource[procDefResources.size()]), dataSource, 
       platformTransactionManager, springAsyncExecutor);

   conf.setDeploymentName(defaultText(activitiProperties.getDeploymentName(), conf.getDeploymentName()));
   conf.setDatabaseSchema(defaultText(activitiProperties.getDatabaseSchema(), conf.getDatabaseSchema()));
   conf.setDatabaseSchemaUpdate(defaultText(activitiProperties.getDatabaseSchemaUpdate(), conf.getDatabaseSchemaUpdate()));
   conf.setDbIdentityUsed(activitiProperties.isDbIdentityUsed());
   conf.setDbHistoryUsed(activitiProperties.isDbHistoryUsed());

   conf.setJobExecutorActivate(activitiProperties.isJobExecutorActivate());
   conf.setAsyncExecutorEnabled(activitiProperties.isAsyncExecutorEnabled());
   conf.setAsyncExecutorActivate(activitiProperties.isAsyncExecutorActivate());
   
   conf.setMailServerHost(activitiProperties.getMailServerHost());
   conf.setMailServerPort(activitiProperties.getMailServerPort());
   conf.setMailServerUsername(activitiProperties.getMailServerUserName());
   conf.setMailServerPassword(activitiProperties.getMailServerPassword());
   conf.setMailServerDefaultFrom(activitiProperties.getMailServerDefaultFrom());
   conf.setMailServerUseSSL(activitiProperties.isMailServerUseSsl());
   conf.setMailServerUseTLS(activitiProperties.isMailServerUseTls());

   conf.setHistoryLevel(activitiProperties.getHistoryLevel());

   if (activitiProperties.getCustomMybatisMappers() != null) {
     conf.setCustomMybatisMappers(getCustomMybatisMapperClasses(activitiProperties.getCustomMybatisMappers()));
   }

   if (activitiProperties.getCustomMybatisXMLMappers() != null) {
     conf.setCustomMybatisXMLMappers(new HashSet<String>(activitiProperties.getCustomMybatisXMLMappers()));
   }
   
   if (processEngineConfigurationConfigurer != null) {
   	processEngineConfigurationConfigurer.configure(conf);
   }

   return conf;
 }

• DataSourceProcessEngineConfiguration的抽象类AbstractProcessEngineAutoConfiguration中有一段代码,表示在spring
  启动时会初始化ProcessEngineFactoryBean，而ProcessEngineFactoryBean是一个FactoryBean，所以在初始化之后会调用getObject方法来真正的加载bean

 @Bean
  public ProcessEngineFactoryBean processEngine(SpringProcessEngineConfiguration configuration) throws Exception {
    return super.springProcessEngineBean(configuration);
  }

getObject()方法代码如下，在这里会调用前面初始化的流程引擎，并调用buildProcessEngine方法

    public ProcessEngine getObject() throws Exception {
        configureExpressionManager();
        configureExternallyManagedTransactions();

        if (processEngineConfiguration.getBeans() == null) {
            processEngineConfiguration.setBeans(new SpringBeanFactoryProxyMap(applicationContext));
        }

        this.processEngine = processEngineConfiguration.buildProcessEngine();
        return this.processEngine;
    }

• 接下来调用配置的buildProcessEngine方法

org.activiti.spring.SpringProcessEngineConfiguration
    public ProcessEngine buildProcessEngine() {
        //2.2.1调用父类的buildProcessEngine
        ProcessEngine processEngine = super.buildProcessEngine();
        ProcessEngines.setInitialized(true);
        //2.2.2自动发布资源
        autoDeployResources(processEngine);
        return processEngine;
    }

父类的buildProcessEngine

org.activiti.engine.impl.cfg.ProcessEngineConfigurationImpl
    public ProcessEngine buildProcessEngine() {
        this.init();
        return new ProcessEngineImpl(this);
    }
    //activiti的核心，初始化
  protected void init() {
  	initConfigurators();
  	configuratorsBeforeInit();
    initProcessDiagramGenerator();
    initHistoryLevel();
    initExpressionManager();
    initDataSource();
    initVariableTypes();
    initBeans();
    initFormEngines();
    initFormTypes();
    initScriptingEngines();
    initClock();
    initBusinessCalendarManager();
    initCommandContextFactory();
    initTransactionContextFactory();
    initCommandExecutors();
    initServices();
    initIdGenerator();
    initDeployers();
    initJobHandlers();
    initJobExecutor();
    initAsyncExecutor();
    initTransactionFactory();
    initSqlSessionFactory();
    initSessionFactories();
    initJpa();
    initDelegateInterceptor();
    initEventHandlers();
    initFailedJobCommandFactory();
    initEventDispatcher();
    initProcessValidator();
    initDatabaseEventLogging();
    configuratorsAfterInit();
  }

自动发布资源

    //这里判断配置的文件夹下是否有配置的bpmn或者xml文件，如果有，则发布资源，这里还用到了策略模式，总共有三种模式，spring默认的是DefaultAutoDeploymentStrategy
    protected void autoDeployResources(ProcessEngine processEngine) {
        if (deploymentResources != null && deploymentResources.length > 0) {
            final AutoDeploymentStrategy strategy = getAutoDeploymentStrategy(deploymentMode);
            strategy.deployResources(deploymentName, deploymentResources, processEngine.getRepositoryService());
        }
    }
    //走进DefaultAutoDeploymentStrategy的deployResources，这里没什么逻辑，就是获取文件流，然后统一调用deploy方法
    public void deployResources(final String deploymentNameHint, final Resource[] resources, final RepositoryService repositoryService) {

        // Create a single deployment for all resources using the name hint as the
        // literal name
         //通过enableDuplicateFiltering将isDuplicateFilterEnabled置为true，判断如果与表里最后一条记录的数据相同则不发布
        final DeploymentBuilder deploymentBuilder = repositoryService.createDeployment().enableDuplicateFiltering().name(deploymentNameHint);

        for (final Resource resource : resources) {
            final String resourceName = determineResourceName(resource);

            try {
                if (resourceName.endsWith(".bar") || resourceName.endsWith(".zip") || resourceName.endsWith(".jar")) {
                    deploymentBuilder.addZipInputStream(new ZipInputStream(resource.getInputStream()));
                } else {
                    deploymentBuilder.addInputStream(resourceName, resource.getInputStream());
                }
            } catch (IOException e) {
                throw new ActivitiException("couldn't auto deploy resource '" + resource + "': " + e.getMessage(), e);
            }
        }

        deploymentBuilder.deploy();

    }
    //走进DeploymentBuilder的deploy的方法，调用了repositoryService的deploy方法
    public Deployment deploy() {
        return repositoryService.deploy(this);
    }
    //repositoryService的deploy方法是熟悉的命令模式，我们可以知道主要逻辑就在DeployCmd的execute方法中
    public Deployment deploy(DeploymentBuilderImpl deploymentBuilder) {
        return commandExecutor.execute(new DeployCmd<Deployment>(deploymentBuilder));
    }
    public Deployment execute(CommandContext commandContext) {
        DeploymentEntity deployment = deploymentBuilder.getDeployment();
        
        deployment.setDeploymentTime(commandContext.getProcessEngineConfiguration().getClock().getCurrentTime());
        //这里isDuplicateFilterEnabled是true
        if ( deploymentBuilder.isDuplicateFilterEnabled() ) {
            
            List<Deployment> existingDeployments = new ArrayList<Deployment>();
             //跟租户相关的逻辑
          if (deployment.getTenantId() == null || ProcessEngineConfiguration.NO_TENANT_ID.equals(deployment.getTenantId())) {
            DeploymentEntity existingDeployment = commandContext
                     .getDeploymentEntityManager()
                     .findLatestDeploymentByName(deployment.getName());
            if (existingDeployment != null) {
                existingDeployments.add(existingDeployment);
            }
          } else {
            //没有租户则通过查询表ACT_RE_DEPLOYMENT来获取最后发布的数据
             List<Deployment> deploymentList = commandContext
                 .getProcessEngineConfiguration().getRepositoryService()
                 .createDeploymentQuery()
                 .deploymentName(deployment.getName())
                 .deploymentTenantId(deployment.getTenantId())
                 .orderByDeploymentId().desc().list();
             
             if (!deploymentList.isEmpty()) {
                 existingDeployments.addAll(deploymentList);
             }
          }
                
          DeploymentEntity existingDeployment = null;
          if(!existingDeployments.isEmpty()) {
            //这里还是只取existingDeployments中的第一条数据
            existingDeployment = (DeploymentEntity) existingDeployments.get(0);
          }
            //2.2.2.1重点，判断是否有改动，如果没有改动则直接返回
          if ( (existingDeployment!=null)
               && !deploymentsDiffer(deployment, existingDeployment)) {
            return existingDeployment;
          }
        }
        //如果判断文件流有改动，则发布
        deployment.setNew(true);

        // Save the data,在ACT_RE_DEPLOYMENT表中插入一条数据
        commandContext
          .getDeploymentEntityManager()
          .insertDeployment(deployment);
         //发布创建Entity事件
        if (commandContext.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
            commandContext.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 
        // 2.2.2.2发布资源
        commandContext
          .getProcessEngineConfiguration()
          .getDeploymentManager()
          .deploy(deployment, deploymentSettings);
        
        if (deploymentBuilder.getProcessDefinitionsActivationDate() != null) {
          scheduleProcessDefinitionActivation(commandContext, deployment);
        }
        
        if (commandContext.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
            commandContext.getProcessEngineConfiguration().getEventDispatcher().dispatchEvent(
                    ActivitiEventBuilder.createEntityEvent(ActivitiEventType.ENTITY_INITIALIZED, deployment));
        }
        
        return deployment;
    }

判断流程文件是否有改动

    protected boolean deploymentsDiffer(DeploymentEntity deployment, DeploymentEntity saved) {
        //调用getResources方法获取对应的资源，如果不为空，则进行比对
        if(deployment.getResources() == null || saved.getResources() == null) {
          return true;
        }
        
        Map<String, ResourceEntity> resources = deployment.getResources();
        Map<String, ResourceEntity> savedResources = saved.getResources();
        
        for (String resourceName: resources.keySet()) {
          ResourceEntity savedResource = savedResources.get(resourceName);
          
          if(savedResource == null) return true;
          
          if(!savedResource.isGenerated()) {
            ResourceEntity resource = resources.get(resourceName);
            
            byte[] bytes = resource.getBytes();
            byte[] savedBytes = savedResource.getBytes();
            if (!Arrays.equals(bytes, savedBytes)) {
              return true;
            }
          }
        }
        return false;
    }
    //看下getResources方法，就是根据传入的deployId来查询ACT_GE_BYTEARRAY表记录，然后构造map返回
    public Map<String, ResourceEntity> getResources() {
        if (resources==null && id!=null) {
          List<ResourceEntity> resourcesList = Context
            .getCommandContext()
            .getResourceEntityManager()
            .findResourcesByDeploymentId(id);
          resources = new HashMap<String, ResourceEntity>();
          for (ResourceEntity resource: resourcesList) {
            resources.put(resource.getName(), resource);
          }
        }
        return resources;
    }

发布资源文件

    public void deploy(DeploymentEntity deployment, Map<String, Object> deploymentSettings) {
    //deployers就只有一个BpmnDeployer，似乎还有另一个RulesDeployer？可以在其他地方再看看用法
        for (Deployer deployer: deployers) {
          deployer.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();
        Map<String, BpmnModel> bpmnModelMap = new HashMap<String, BpmnModel>();
        
        final ProcessEngineConfigurationImpl processEngineConfiguration = Context.getProcessEngineConfiguration();
        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 bpmnParse = bpmnParser
              .createParse()
              .sourceInputStream(inputStream)
              .setSourceSystemId(resourceName)
              .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));
                }
                
            } else {
                // On redeploy, we assume it is validated at the first deploy
                bpmnParse.setValidateSchema(false);
                bpmnParse.setValidateProcess(false);
            }
            //校验bpmn文件
            bpmnParse.execute();
            
            for (ProcessDefinitionEntity processDefinition: bpmnParse.getProcessDefinitions()) {
              processDefinition.setResourceName(resourceName);
              //这里租户id是空字符串
              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 (processEngineConfiguration.isCreateDiagramOnDeploy() &&
                      diagramResourceName==null && processDefinition.isGraphicalNotationDefined()) {
                  try {
                      //生成图片流
                      byte[] diagramBytes = IoUtil.readInputStream(processEngineConfiguration.
                        getProcessDiagramGenerator().generateDiagram(bpmnParse.getBpmnModel(), "png", processEngineConfiguration.getActivityFontName(),
                            processEngineConfiguration.getLabelFontName(),processEngineConfiguration.getAnnotationFontName(), processEngineConfiguration.getClassLoader()), null);
                      diagramResourceName = getProcessImageResourceName(resourceName, processDefinition.getKey(), "png");
                      //保存ACT_GE_BYTEARRAY数据
                      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);
              bpmnModelMap.put(processDefinition.getKey(), bpmnParse.getBpmnModel());
            }
          }
        }
        
        // 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 '"+ processDefinition.getKey() +"' (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;
            //查询ACT_RE_PROCDEF
            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);
            
            removeExistingMessageEventSubscriptions(processDefinition, latestProcessDefinition);
            addMessageEventSubscriptions(processDefinition);
            
            removeExistingSignalEventSubScription(processDefinition, latestProcessDefinition);
            addSignalEventSubscriptions(processDefinition);
            //插入表ACT_RE_PROCDEF
            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
          DeploymentManager deploymentManager = processEngineConfiguration.getDeploymentManager();
          deploymentManager.getProcessDefinitionCache().add(processDefinition.getId(), processDefinition);
          addDefinitionInfoToCache(processDefinition, processEngineConfiguration, commandContext);
          
          // Add to deployment for further usage
          deployment.addDeployedArtifact(processDefinition);
          
          createLocalizationValues(processDefinition.getId(), bpmnModelMap.get(processDefinition.getKey()).getProcessById(processDefinition.getKey()));
        }
    }

总结

• spring启动时会加载流程引擎类并调用buildProcessEngine方法
• 发布时会拿文件夹下的所有文件与表中的byte数组逐个对比，有一个不一样则全量发布
• 发布时会记录
  act_re_deployment：本次发布的信息,
  act_ge_bytearray：本次发布的文件流包括bpmn文件及生成的图片流,
  act_re_procdef：记录发布的路径及版本号信息