184 - 异步执行器设计和配置

Posted 2022-12-04 分享牛

1、异步执行器设计

存在两种类型的作业：定时器（例如属于用户任务上的边界事件的定时器）和异步延续（属于具有流动性：异步=“真”属性的服务任务）。

定时器是最容易解释的：它们在ACT_RU_TIMER_JOB表中持有一定的到期日期。异步执行程序中有一个线程定期检查是否有新的定时器应该触发（换句话说，截止日期是在当前时间之前）。发生这种情况时，定时器将被删除，并创建一个异步作业。

在执行流程实例步骤期间（即在进行某个API调用期间），将异步作业插入到数据库中。如果异步执行程序对当前的Flowable引擎有效，则异步作业实际上已经被锁定。这意味着作业条目被插入到ACT_RU_JOB表中，并且将有一个锁拥有者和一个锁过期时间组。一个成功提交API调用的事务监听器触发同一个引擎的异步执行器来执行作业（所以数据保证在数据库中）。为此，异步执行程序具有一个可配置的线程池，线程将从该线程池执行该作业并异步地继续该过程。如果Flowable引擎没有启用异步执行程序，则将异步作业插入到ACT_RU_JOB表中而不被锁定。

类似于检查新定时器的线程，异步执行程序有一个线程获取新的异步作业。这些是表中没有锁定的作业。此线程将锁定当前Flowable引擎的这些作业并将其传递给异步执行程序。

执行作业的线程池使用内存队列从中进行作业。当这个队列已满（这是可配置的），作业将被解锁并重新插入到它的表中。这样，其他的异步执行者可以代替它。

如果在作业执行期间发生异常，则异步作业将转换为具有截止日期的计时器作业。之后，它将像普通的定时器工作一样被选中，并再次成为一个异步工作，很快就会被重新尝试。如果某个作业重试了（可配置的）次数并继续失败，则认为该作业已经死机并移至ACT_RU_DEADLETTER_JOB。该死信概念被广泛应用于各种其他系统。管理员现在需要检查失败作业的例外情况，并决定最佳的行动方式。

流程定义和流程实例可以暂停。将与这些定义或实例相关的挂起作业放入ACT_RU_SUSPENDED_JOB表中，以确保获取作业的查询在where子句中具有尽可能少的条件。

有一点从上面可以清楚的看出：对于那些熟悉这个工作/异步执行者的旧实现的人来说，主要目标是允许获取查询尽可能简单。在过去的（V6之前），用于所有作业类型/状态，这使一个表中的其中条件的大，因为它照顾到所有用例。现在这个问题已经解决，我们的基准已经证明这个新的设计提供了更好的性能，并且更具可扩展性。

2、异步执行器配置

异步执行程序是一个高度可配置的组件。总是建议查看异步执行程序的默认设置，并验证它们是否符合您的进程的要求。

或者，可以扩展默认实现，或者将org.flowable.engine.impl.asyncexecutor.AsyncExecutor接口替换为您自己的实现。

流程引擎配置通过setter提供以下属性：
表1.异步执行程序配置选项

名称	默认值	描述
asyncExecutorThreadPoolQueueSize	100	执行作业的队列的大小在被线程池中的线程实际执行之前被放置
asyncExecutorCorePoolSize	2	线程池中用于执行作业的最小线程数。
asyncExecutorMaxPoolSize	10	线程池中为作业执行而创建的最大线程数。
asyncExecutorThreadKeepAliveTime	5000	用于作业执行的线程在被销毁前必须保持活动状态（以毫秒为单位）。设置> 0需要资源，但是在执行多个作业的情况下，它总是避免创建新的线程。如果为0，线程在用于执行作业后将被销毁。
asyncExecutorNumberOfRetries	3	将作业转移到死书表之前将被重试的次数。
asyncExecutorMaxTimerJobsPerAcquisition	1	在一个查询中获取的计时器作业的数量。默认值是1，因为这降低了乐观锁定异常的可能性。较大的值可以执行得更好，但在不同引擎之间发生乐观锁定异常的机会也会变得更大。
asyncExecutorMaxAsyncJobsDuePerAcquisition	1	在一个查询中获取的异步作业的数量。默认值是1，因为这降低了乐观锁定异常的可能性。较大的值可以执行得更好，但在不同引擎之间发生乐观锁定异常的机会也会变得更大。
asyncExecutorDefaultTimerJobAcquireWaitTime	10000	定时器获取线程的时间（以毫秒为单位）将等待执行下一个查询。发生这种情况时没有找到新的计时器作业，或者获取的计时器作业比asyncExecutorMaxTimerJobsPerAcquisition中设置的时间少。
asyncExecutorDefaultAsyncJobAcquireWaitTime	10000	异步作业获取线程的时间（以毫秒为单位）将等待执行下一个查询。发生这种情况时没有找到新的异步作业，或者获取的异步作业少于asyncExecutorMaxAsyncJobsDuePerAcquisition中的设置。
asyncExecutorDefaultQueueSizeFullWaitTime	0	当内部作业队列已满时，异步作业（包括定时器和异步继续）获取线程的时间（以毫秒为单位）将等待执行下一个查询。默认情况下设置为0（为了向后兼容）。将此属性设置为更高的值可以使异步执行程序有希望清除队列。
asyncExecutorTimerLockTimeInMillis	5分钟	计时器作业的时间量（以毫秒为单位）被异步执行程序获取时锁定。在这段时间内，没有其他异步执行者会尝试获取和锁定这个工
作。
asyncExecutorAsyncJobLockTimeInMillis	5分钟	异步作业被异步执行程序获取的时间量（以毫秒为单位）被锁定。在这段时间内，没有其他异步执行者会尝试获取和锁定这个工作。
asyncExecutorSecondsToWaitOnShutdown	60	请求在执行程序（或进程引擎）关闭时等待
asyncExecutorResetExpiredJobsInterval	60秒	过期作业的两次连续检查之间的时间量（以毫秒为单位）。过期的作业是锁定的作业（锁定所有者+时间是由某个执行者编写的，但是作业从未完成）。在这种检查过程中，过期的作业将再次可用，这意味着锁拥有者和锁定时间将被移除。其他执行者现在将能够拿起它。如果锁定时间在当前日期之前，则认为工作已过期。
asyncExecutorResetExpiredJobsPageSize	3	由异步执行程序的重置过期线程立即提取的作业数量。

上面文章来自盘古BPM研究院：http://vue.pangubpm.com/
文章翻译提交：https://github.com/qiudaoke/flowable-userguide
了解更多文章可以关注微信公众号：