【问题标题】:Activiti 6 sometime waits 5 minutes before starting async process instanceActiviti 6 有时会在启动异步流程实例之前等待 5 分钟
【发布时间】:2019-07-23 23:22:23
【问题描述】:

当我发出进程启动命令时,Activiti 6.0.0(带有spring boot)保持异步进程并且不会立即启动它: (发出此命令时有一个打开的 JPA 事务,实体作为变量传递给进程)

ProcessInstance processInstance = runtimeService.startProcessInstanceByKey(processId, variableMap);

5 分钟后运行它没有任何错误。

日志:

15:17:01.158 DEBUG [http-nio-8080-exec-5] c.m.i.e.p.input.impl.ImportingPortImpl   : Starting workflow process... [workflowName=batch-main]
<5 MINUTE TEA BREAK>
15:22:10.677 DEBUG [SimpleAsyncTaskExecutor-6] c.m.i.e.i.c.analizers.ContentAnalizers   : Task really started.. 

这发生在大约 30% 的情况下,在其他情况下它会立即开始,所以这很奇怪。

如何解决此问题以在每种情况下都立即运行,而无需等待 5 分钟超时?

更详细的日志:

16:53:14.136 DEBUG [http-nio-8080-exec-2] c.m.i.e.r.a.i.IntellivectorImportAdapter : Getting content stream from multipart [multipartIndex=0;fileName=IMPORTED-4e578b1a-e7f6-4b5a-91d0-9e9a40f3328e.tiff;contentType=image/tiff]

16:53:14.137 DEBUG [http-nio-8080-exec-2] o.h.e.t.internal.TransactionImpl         : begin
16:53:14.138  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : select count(RES.ID_) from ACT_RE_PROCDEF RES WHERE RES.KEY_ = 'batch-main' 
16:53:14.142  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : select nextval ('iv_batch_id_seq') 
16:53:14.144  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : select nextval ('iv_document_id_seq') 
16:53:14.147  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : insert into iv_batch (created_date, id) values ('11/28/2017 16:53:14.142', 1245) 
16:53:14.148  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : insert into iv_document (batch_id, created_date, id) values (1245, '11/28/2017 16:53:14.144', 1272) 
16:53:14.150  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : insert into iv_document_contents (mime_type, storage_id, document_id, phase) values ('image/tiff', 'c74d06ef-3c92-4c22-8238-7e45e80d53e3', 1272, 'IMPORTED') 
16:53:14.150  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : update iv_document_contents set index_=0 where document_id=1272 and phase='IMPORTED' 

16:53:14.151 DEBUG [http-nio-8080-exec-2] c.m.i.e.p.input.impl.ImportingPortImpl   : Imported a batch, starting workflow... [batchId=1245;documentCount=1;workflowName=batch-main]
16:53:14.151  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : select * from ACT_RE_PROCDEF where KEY_ = 'batch-main' and (TENANT_ID_ = '' or TENANT_ID_ is null) and VERSION_ = (select max(VERSION_) from ACT_RE_PROCDEF where KEY_ = 'batch-main' and (TENANT_ID_ = '' or TENANT_ID_ is null)) 
16:53:14.153  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : insert into ACT_HI_VARINST (ID_, PROC_INST_ID_, EXECUTION_ID_, TASK_ID_, NAME_, REV_, VAR_TYPE_, BYTEARRAY_ID_, DOUBLE_, LONG_ , TEXT_, TEXT2_, CREATE_TIME_, LAST_UPDATED_TIME_) 
values ( '5048', '5047', '5047', NULL, 'batchId', 0, 'long', NULL, NULL, 1245, '1245', NULL, '11/28/2017 16:53:14.153', '11/28/2017 16:53:14.153' ) 

16:53:14.154  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : insert into ACT_HI_PROCINST ( ID_, PROC_INST_ID_, BUSINESS_KEY_, PROC_DEF_ID_, START_TIME_, END_TIME_, DURATION_, START_USER_ID_, START_ACT_ID_, END_ACT_ID_, SUPER_PROCESS_INSTANCE_ID_, DELETE_REASON_, TENANT_ID_, NAME_ ) 
values ( '5047', '5047', NULL, 'batch-main:3:5021', '11/28/2017 16:53:14.152', NULL, NULL, NULL, 'theStart', NULL, NULL, NULL, '', NULL ) 
16:53:14.156  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : insert into ACT_RU_EXECUTION (ID_, REV_, PROC_INST_ID_, BUSINESS_KEY_, PROC_DEF_ID_, ACT_ID_, IS_ACTIVE_, IS_CONCURRENT_, IS_SCOPE_,IS_EVENT_SCOPE_, IS_MI_ROOT_, PARENT_ID_, SUPER_EXEC_, ROOT_PROC_INST_ID_, SUSPENSION_STATE_, TENANT_ID_, NAME_, START_TIME_, START_USER_ID_, IS_COUNT_ENABLED_, EVT_SUBSCR_COUNT_, TASK_COUNT_, JOB_COUNT_, TIMER_JOB_COUNT_, SUSP_JOB_COUNT_, DEADLETTER_JOB_COUNT_, VAR_COUNT_, ID_LINK_COUNT_) 
values ('5047', 1, '5047', NULL, 'batch-main:3:5021', NULL, 1, 0, 1, 0, 0, NULL, NULL, '5047', 1, '', NULL, '11/28/2017 16:53:14.152', NULL, 0, 0, 0, 0, 0, 0, 0, 0, 0) , ('5049', 1, '5047', NULL, 'batch-main:3:5021', 'theStart', 1, 0, 0, 0, 0, '5047', NULL, '5047', 1, '', NULL, '11/28/2017 16:53:14.153', NULL, 0, 0, 0, 0, 0, 0, 0, 0, 0) 
16:53:14.157  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : insert into ACT_RU_VARIABLE (ID_, REV_, TYPE_, NAME_, PROC_INST_ID_, EXECUTION_ID_, TASK_ID_, BYTEARRAY_ID_, DOUBLE_, LONG_ , TEXT_, TEXT2_) 
values ( '5048', 1, 'long', 'batchId', '5047', '5047', NULL, NULL, NULL, 1245, '1245', NULL ) 
16:53:14.159  INFO [http-nio-8080-exec-2] jdbc.sqlonly                             : insert into ACT_RU_JOB ( ID_, REV_, TYPE_, LOCK_OWNER_, LOCK_EXP_TIME_, EXCLUSIVE_, EXECUTION_ID_, PROCESS_INSTANCE_ID_, PROC_DEF_ID_, RETRIES_, EXCEPTION_STACK_ID_, EXCEPTION_MSG_, DUEDATE_, REPEAT_, HANDLER_TYPE_, HANDLER_CFG_, TENANT_ID_) 
values ('5050', 1, 'message', 'e53309f8-8780-4849-be82-c9acfbffa338', '11/28/2017 16:58:14.153', 1, '5049', '5047', 'batch-main:3:5021', 3, NULL, NULL, NULL, NULL, 'async-continuation', NULL, '' ) 

    16:53:14.160 DEBUG [http-nio-8080-exec-2] o.h.e.t.internal.TransactionImpl         : begin
    16:53:14.160 DEBUG [http-nio-8080-exec-2] o.h.e.t.internal.TransactionImpl         : committing
16:53:14.160 DEBUG [http-nio-8080-exec-2] o.h.e.t.internal.TransactionImpl         : committing


16:53:14.161 DEBUG [SimpleAsyncTaskExecutor-2] o.h.e.t.internal.TransactionImpl         : begin
16:53:14.161  INFO [SimpleAsyncTaskExecutor-2] jdbc.sqlonly                             : select * from ACT_RU_EXECUTION where ID_ = '5049' 

16:53:14.161 DEBUG [SimpleAsyncTaskExecutor-2] o.h.e.t.internal.TransactionImpl         : committing
16:53:14.162 DEBUG [SimpleAsyncTaskExecutor-2] o.h.e.t.internal.TransactionImpl         : begin
16:53:14.162  INFO [SimpleAsyncTaskExecutor-2] jdbc.sqlonly                             : select * from ACT_RU_JOB where ID_ = '5050' 

16:53:14.163 DEBUG [SimpleAsyncTaskExecutor-2] o.h.e.t.internal.TransactionImpl         : committing
16:53:14.163 DEBUG [SimpleAsyncTaskExecutor-2] o.h.e.t.internal.TransactionImpl         : begin
16:53:14.163  INFO [SimpleAsyncTaskExecutor-2] jdbc.sqlonly                             : select * from ACT_RU_EXECUTION where ID_ = '5047' 

16:53:14.164 DEBUG [SimpleAsyncTaskExecutor-2] o.h.e.t.internal.TransactionImpl         : committing
16:53:22.492 DEBUG [activiti-acquire-async-jobs] o.h.e.t.internal.TransactionImpl         : begin
16:53:22.493  INFO [activiti-acquire-async-jobs] jdbc.sqlonly                             : select RES.* from ACT_RU_JOB RES where LOCK_EXP_TIME_ is null LIMIT 1 OFFSET 0 

16:53:22.493 DEBUG [activiti-acquire-timer-jobs] o.h.e.t.internal.TransactionImpl         : begin
16:53:22.493 DEBUG [activiti-acquire-async-jobs] o.h.e.t.internal.TransactionImpl         : committing
16:53:22.493  INFO [activiti-acquire-timer-jobs] jdbc.sqlonly                             : select RES.* from ACT_RU_TIMER_JOB RES where DUEDATE_ <= '11/28/2017 16:53:22.493' and LOCK_OWNER_ is null LIMIT 1 OFFSET 0 

16:53:22.495 DEBUG [activiti-acquire-timer-jobs] o.h.e.t.internal.TransactionImpl         : committing
16:53:22.495 DEBUG [activiti-acquire-timer-jobs] o.h.e.t.internal.TransactionImpl         : begin
16:53:22.495 DEBUG [activiti-acquire-timer-jobs] o.h.e.t.internal.TransactionImpl         : committing

<NOT STARTED.... >

露天线程:Async Job Executor only executes job after lock expiration

【问题讨论】:

  • 我对 Activiti 不熟悉,但一种解释是,Activiti 的主线程使用的是常用的每 5 分钟唤醒一次的模式(可能使用@Scheduled)来检查是否有任何需要运行的东西,而不是立即触发任务。或者它可能正在等待另一个当前正在运行的任务完成,然后再触发您的任务,因为它被配置为顺序执行而不是并行执行。
  • 不,没有其他任务,在其他情况下它立即运行,而不是在 0-5 分钟之间的随机超时。对我来说,超时似乎是任务超时,对于长时间运行的任务,假设它已损坏并尝试再次运行,因此 Activiti 似乎认为该任务正在运行,并在 5 分钟后再次尝试。
  • 您可以发布您的流程 xml 架构吗?你能在启动过程后确认它真的没有立即启动吗?喜欢Log.info("Number of process instances: " + runtimeService.createProcessInstanceQuery().count());
  • 还有16:53:14.151 DEBUG [http-nio-8080-exec-2] c.m.i.e.p.input.impl.ImportingPortImpl : Imported a batch, starting workflow... 这是什么时候打印的? (在进程启动之前或之后)您是否在启动进程之前执行任何批处理查询?
  • @CrazySabbath:之前。是的,在此之前启动时有重新启动事件,这会再次触发一些等待任务以将某些内容加载到运行时业务组件,但这些似乎在单独的事务中更早地正确完成。

标签: java activiti


【解决方案1】:

我已将其缩小到此问题:选中“异步”且未选中“独占”的任务会导致延迟。我创建了一个展示行为的工作流。五个任务,每个任务都休眠一秒钟,并设置了“异步”和“独占”标志的各种组合。在下面的日志 sn-p 中,请注意任务“scripttask04_async”之前的 5 分钟延迟以及 ExecuteAsyncJobCmd 记录的错误消息。

engine 17:25:48.337 DEBUG NfExecutionListener            execution id: 60065, received event: start, task id: null, parent id: null, variables {admin=null}, local variables {admin=VariableInstanceEntity[id=60066, name=admin, type=null]}
engine 17:25:48.396 DEBUG ExecuteAsyncJobCmd             Executing async job 60068
engine 17:25:48.407 DEBUG NfExecutionListener            execution id: 60065, received event: start, task id: null, parent id: null, variables {admin=null}, local variables {admin=VariableInstanceEntity[id=60066, name=admin, type=null]}
engine 17:25:48.407 DEBUG ContinueProcessOperation       Executing activityBehavior class org.activiti.engine.impl.bpmn.behavior.NoneStartEventActivityBehavior on activity 'debugHelperStart' with execution 60067
engine 17:25:48.411 DEBUG TakeOutgoingSequenceFlowsOperation Leaving flow node class org.activiti.bpmn.model.StartEvent with id 'debugHelperStart' by following it's 1 outgoing sequenceflow
engine 17:25:48.413 DEBUG ContinueProcessOperation       Sequence flow 'flow1' encountered. Continuing process by following it using execution 60067
engine 17:25:48.414 DEBUG ContinueProcessOperation       Executing activityBehavior class org.activiti.engine.impl.bpmn.behavior.ScriptTaskActivityBehavior on activity 'debugHelper01Task_sync_excl' with execution 60067
 ==========================  ===   ==== before sleep debugHelper01Task_sync_excl
engine 17:25:49.308 INFO  AnnotationConfigApplicationContext Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@40b5162b: startup date [Tue May 01 17:25:49 UTC 2018]; parent: org.springframework.boot.context.embedded.AnnotationConfigEmbeddedWebApplicationContext@3439f68d
engine 17:25:49.407 INFO  AutowiredAnnotationBeanPostProcessor JSR-330 'javax.inject.Inject' annotation found and supported for autowiring
engine 17:25:49.843 INFO  ChainedDynamicProperty         Flipping property: localhost.ribbon.ActiveConnectionsLimit to use NEXT property: niws.loadbalancer.availabilityFilteringRule.activeConnectionsLimit = 2147483647
 ==========================  ===   ==== after sleep debugHelper01Task_sync_excl
engine 17:25:49.851 DEBUG TakeOutgoingSequenceFlowsOperation Leaving flow node class org.activiti.bpmn.model.ScriptTask with id 'debugHelper01Task_sync_excl' by following it's 1 outgoing sequenceflow
engine 17:25:49.851 DEBUG ContinueProcessOperation       Sequence flow 'flow2' encountered. Continuing process by following it using execution 60067
engine 17:25:49.960 INFO  BaseLoadBalancer               Client: localhost instantiated a LoadBalancer: DynamicServerListLoadBalancer:{NFLoadBalancer:name=localhost,current list of Servers=[],Load balancer stats=Zone stats: {},Server stats: []}ServerList:null
engine 17:25:49.974 INFO  DynamicServerListLoadBalancer  Using serverListUpdater PollingServerListUpdater
engine 17:25:49.986 INFO  DynamicServerListLoadBalancer  DynamicServerListLoadBalancer for client localhost initialized: DynamicServerListLoadBalancer:{NFLoadBalancer:name=localhost,current list of Servers=[],Load balancer stats=Zone stats: {},Server stats: []}ServerList:com.netflix.loadbalancer.ConfigurationBasedServerList@50af6398
engine 17:25:59.119 DEBUG ExecuteAsyncJobCmd             Executing async job 60072
engine 17:25:59.122 DEBUG ContinueProcessOperation       Executing activityBehavior class org.activiti.engine.impl.bpmn.behavior.ScriptTaskActivityBehavior on activity 'scripttask02_async_excl' with execution 60067
 ==========================  ===   ==== before sleep scripttask02_async_excl
 ==========================  ===   ==== after sleep scripttask02_async_excl
engine 17:26:00.132 DEBUG TakeOutgoingSequenceFlowsOperation Leaving flow node class org.activiti.bpmn.model.ScriptTask with id 'scripttask02_async_excl' by following it's 1 outgoing sequenceflow
engine 17:26:00.132 DEBUG ContinueProcessOperation       Sequence flow 'flow3' encountered. Continuing process by following it using execution 60067
engine 17:26:00.132 DEBUG ContinueProcessOperation       Executing activityBehavior class org.activiti.engine.impl.bpmn.behavior.ScriptTaskActivityBehavior on activity 'scripttask03_sync' with execution 60067
 ==========================  ===   ==== before sleep scripttask03_sync
 ==========================  ===   ==== after sleep scripttask03_sync
engine 17:26:01.136 DEBUG TakeOutgoingSequenceFlowsOperation Leaving flow node class org.activiti.bpmn.model.ScriptTask with id 'scripttask03_sync' by following it's 1 outgoing sequenceflow
engine 17:26:01.136 DEBUG ContinueProcessOperation       Sequence flow 'flow4' encountered. Continuing process by following it using execution 60067
engine 17:26:01.148 DEBUG ExecuteAsyncJobCmd             Job does not exist anymore and will not be executed. It has most likely been deleted as part of another concurrent part of the process instance.
engine 17:31:39.388 DEBUG ExecuteAsyncJobCmd             Executing async job 60076
engine 17:31:39.392 DEBUG ContinueProcessOperation       Executing activityBehavior class org.activiti.engine.impl.bpmn.behavior.ScriptTaskActivityBehavior on activity 'scripttask04_async' with execution 60067
 ==========================  ===   ==== before sleep scripttask04_async
 ==========================  ===   ==== after sleep scripttask04_async
engine 17:31:40.401 DEBUG TakeOutgoingSequenceFlowsOperation Leaving flow node class org.activiti.bpmn.model.ScriptTask with id 'scripttask04_async' by following it's 1 outgoing sequenceflow
engine 17:31:40.402 DEBUG ContinueProcessOperation       Sequence flow 'flow5' encountered. Continuing process by following it using execution 60067
engine 17:31:40.422 DEBUG ExecuteAsyncJobCmd             Executing async job 60078
engine 17:31:40.425 DEBUG ContinueProcessOperation       Executing activityBehavior class org.activiti.engine.impl.bpmn.behavior.ScriptTaskActivityBehavior on activity 'scripttask05_async_excl' with execution 60067
 ==========================  ===   ==== before sleep scripttask05_async_excl
 ==========================  ===   ==== after sleep scripttask05_async_excl
engine 17:31:41.440 DEBUG TakeOutgoingSequenceFlowsOperation Leaving flow node class org.activiti.bpmn.model.ScriptTask with id 'scripttask05_async_excl' by following it's 1 outgoing sequenceflow
engine 17:31:41.441 DEBUG ContinueProcessOperation       Sequence flow 'flow6' encountered. Continuing process by following it using execution 60067
engine 17:31:41.441 DEBUG ContinueProcessOperation       Executing activityBehavior class org.activiti.engine.impl.bpmn.behavior.NoneEndEventActivityBehavior on activity 'debugHelperEnd' with execution 60067
engine 17:31:41.441 DEBUG TakeOutgoingSequenceFlowsOperation Leaving flow node class org.activiti.bpmn.model.EndEvent with id 'debugHelperEnd' by following it's 0 outgoing sequenceflow
engine 17:31:41.441 DEBUG TakeOutgoingSequenceFlowsOperation No outgoing sequence flow found for flow node 'debugHelperEnd'.
engine 17:31:41.476 DEBUG NfExecutionListener            execution id: 60065, received event: end, task id: null, parent id: null, variables {admin=null}, local variables {admin=VariableInstanceEntity[id=60066, name=admin, type=null]}

【讨论】:

    【解决方案2】:

    我怀疑与ExecuteAsyncJobCmd 结合的未结交易是导致此问题的原因。来自命令的代码库:

    // We need to refetch the job, as it could have been deleted by another concurrent job
    // For exampel: an embedded subprocess with a couple of async tasks and a timer on the boundary of the subprocess
    // when the timer fires, all executions and thus also the jobs inside of the embedded subprocess are destroyed.
    // However, the async task jobs could already have been fetched and put in the queue.... while in reality they have been deleted. 
    // A refetch is thus needed here to be sure that it exists for this transaction.
    
    Job job = commandContext.getJobEntityManager().findById(jobId);
    if (job == null) {
      log.debug("Job does not exist anymore and will not be executed. It has most likely been deleted "
          + "as part of another concurrent part of the process instance.");
      return null;
    }
    

    这是我们在使用 Activiti Spring Boot Starter 中提供的默认流程引擎配置测试内存数据库和异步作业时遇到的确切行为。

    【讨论】:

      【解决方案3】:

      我有同样的问题,我想知道为什么会这样,所以我研究了 Activiti 6.0 的源代码。

      事实证明这是一个并发错误,它与事务处理有关,并且在您使用 activiti-spring 时会特别发生。

      在 GitHub 上查看问题:https://github.com/Activiti/Activiti/issues/2666

      具体来说,当一个作业导致另一个作业被创建时,Activiti 的异步执行器有可能在将作业实体插入数据库的事务提交之前开始执行作业。由于命令的执行方式和事务包装完成的方式,这特别发生在 activiti-spring 中。

      我目前没有很好的解决方法(避免这种情况的一种方法是不使用 activiti-spring,但如果您需要在应用程序中集成 Spring,那么这是不可行的)。

      【讨论】:

        【解决方案4】:

        正如 Jesper 所说,问题的根本原因是在事务提交之前将作业提交给执行器服务,方法 commandContext.getJobEntityManager().findById(jobId); 会返回空值:

        Job job = commandContext.getJobEntityManager().findById(jobId);
        if (job == null) {
          log.debug("Job does not exist anymore and will not be executed. It has most likely been deleted "
              + "as part of another concurrent part of the process instance.");
          return null;
        }
        

        基于消息队列的异步执行器不会出现此问题,因为消息是在提交后发送到队列的。

        【讨论】:

          猜你喜欢
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          • 2016-04-09
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          • 2017-11-14
          • 2019-07-13
          相关资源
          最近更新 更多