camunda框架学习（十二）外部任务系列

目前，外部任务仅camunda框架才有。

一、外部任务抓取锁定使用

    <bpmn:serviceTask id="Activity_05ut8e6" name="外部任务1" camunda:type="external" camunda:topic="topic1" camunda:taskPriority="50"><bpmn:incoming>Flow_1p9ui82</bpmn:incoming><bpmn:outgoing>Flow_0l61ok8</bpmn:outgoing></bpmn:serviceTask><bpmn:serviceTask id="Activity_0y8yfpp" name="外部任务2" camunda:type="external" camunda:topic="topic2" camunda:taskPriority="100"><bpmn:incoming>Flow_0l61ok8</bpmn:incoming><bpmn:outgoing>Flow_0j1eimn</bpmn:outgoing></bpmn:serviceTask>

启动流程时，会向ACT_RU_EXT_TASK插入一条数据

外部任务数据查询：

    /*** select distinct RES.REV_, RES.ID_, RES.TOPIC_NAME_, RES.WORKER_ID_, RES.LOCK_EXP_TIME_, RES.RETRIES_, RES.ERROR_MSG_, RES.ERROR_DETAILS_ID_, RES.EXECUTION_ID_, RES.PROC_INST_ID_, RES.PROC_DEF_ID_, RES.PROC_DEF_KEY_, RES.ACT_ID_, RES.ACT_INST_ID_, RES.SUSPENSION_STATE_, RES.TENANT_ID_, RES.PRIORITY_, RES.BUSINESS_KEY_ from ( select RES.*, PI.BUSINESS_KEY_* from ACT_RU_EXT_TASK RES left join ACT_RU_EXECUTION PI on RES.PROC_INST_ID_ = PI.ID_ ) RES order by RES.ID_ asc LIMIT ? OFFSET ?*/@Testpublic void createExternalTaskQuery() {List<ExternalTask> list = externalTaskService.createExternalTaskQuery().list();for (ExternalTask externalTask : list) {System.out.println("####" + externalTask);}}

外部服务进行抓取与锁定：

 @Testpublic void fetchAndLock() {int maxTasks = 5;String workerId = "peng1";boolean usePriority = true;int lockDuratio = 1000 * 60 * 10;List<LockedExternalTask> lockedExternalTaskList = externalTaskService.fetchAndLock(maxTasks, workerId, usePriority).topic("topic1", lockDuratio).execute();for (LockedExternalTask lockedExternalTask : lockedExternalTaskList) {System.out.println("#########");System.out.println(lockedExternalTask.getPriority());System.out.println(lockedExternalTask.getId());System.out.println(lockedExternalTask.getTopicName());System.out.println("#########");}}

LOCK_EXP_TIME_要么空，要么小于当前系统时间

二、外部任务优先级说明

    <bpmn:serviceTask id="Activity_05ut8e6" name="外部任务1" camunda:type="external" camunda:topic="topic1" camunda:taskPriority="${priority}"><bpmn:incoming>Flow_1p9ui82</bpmn:incoming><bpmn:outgoing>Flow_0l61ok8</bpmn:outgoing></bpmn:serviceTask><bpmn:serviceTask id="Activity_0y8yfpp" name="外部任务2" camunda:type="external" camunda:topic="topic2"><bpmn:incoming>Flow_0l61ok8</bpmn:incoming><bpmn:outgoing>Flow_0j1eimn</bpmn:outgoing></bpmn:serviceTask>

  <bpmn:process id="topic" isExecutable="true" camunda:taskPriority="100">

    @Testpublic void startProcessInstanceByKey() {Map<String, Object> vars = new HashMap<>();vars.put("priority", 50);runtimeService.startProcessInstanceByKey("topic", vars);}

三、多主题多条件查询

外部任务重复抓取以及锁定:

前提是上一次的锁还没有过期

抓取时获取变量：

@Test
public void startProcessInstanceByKey2() {VariableMap variableMap = Variables.createVariables().putValue("priority", 200).putValue("peng1", "变量1").putValue("peng2", "变量2");runtimeService.startProcessInstanceByKey("topic", variableMap);
}@Test
public void fetchAndLock2() {int maxTasks = 5;String workerId = "peng3";boolean usePriority = true;int lockDuration = 1000 * 10;List<LockedExternalTask> lockedExternalTaskList = externalTaskService.fetchAndLock(maxTasks, workerId, usePriority).topic("topic1", lockDuration).variables("peng1").execute();for (LockedExternalTask lockedExternalTask : lockedExternalTaskList) {System.out.println("#########");System.out.println(lockedExternalTask.getPriority());System.out.println(lockedExternalTask.getId());System.out.println(lockedExternalTask.getTopicName());VariableMap variables = lockedExternalTask.getVariables();System.out.println(variables);System.out.println("#########");}
}

抓取多主题，条件过滤：

@Testpublic void fetchAndLock3() {int maxTasks = 5;String workerId = "peng3";boolean usePriority = true;int lockDuration = 1000 * 10;List<LockedExternalTask> lockedExternalTaskList = externalTaskService.fetchAndLock(maxTasks, workerId, usePriority).topic("topic1", lockDuration).topic("topic2", lockDuration).topic("topic3", lockDuration).processDefinitionIdIn("topic:3:503").execute();for (LockedExternalTask lockedExternalTask : lockedExternalTaskList) {System.out.println("#########");System.out.println(lockedExternalTask.getPriority());System.out.println(lockedExternalTask.getId());System.out.println(lockedExternalTask.getTopicName());VariableMap variables = lockedExternalTask.getVariables();System.out.println(variables);System.out.println("#########");}}

四、外部任务挂起与激活

 @Testpublic void suspendProcessInstanceById() {runtimeService.suspendProcessInstanceById("901");}@Testpublic void activateProcessInstanceById() {runtimeService.activateProcessInstanceById("901");}@Testpublic void fetchAndLock() {int maxTasks = 5;String workerId = "peng3";boolean usePriority = true;int lockDuration = 1000 * 10 * 5;List<LockedExternalTask> lockedExternalTaskList = externalTaskService.fetchAndLock(maxTasks, workerId, usePriority).topic("topic1", lockDuration).execute();for (LockedExternalTask lockedExternalTask : lockedExternalTaskList) {System.out.println("#########");System.out.println(lockedExternalTask.getPriority());System.out.println(lockedExternalTask.getId());System.out.println(lockedExternalTask.getTopicName());System.out.println("#########");}}

五、完成外部任务

@Testpublic void complete() {String externalTaskId = "1504";String workerId = "peng3";externalTaskService.complete(externalTaskId, workerId);}/*** insert into ACT_HI_VARINST ( ID_, PROC_DEF_KEY_, PROC_DEF_ID_, ROOT_PROC_INST_ID_, PROC_INST_ID_, EXECUTION_ID_, ACT_INST_ID_, TENANT_ID_, CASE_DEF_KEY_, CASE_DEF_ID_, CASE_INST_ID_, CASE_EXECUTION_ID_, TASK_ID_, NAME_, REV_, VAR_TYPE_, CREATE_TIME_, REMOVAL_TIME_, BYTEARRAY_ID_, DOUBLE_, LONG_, TEXT_, TEXT2_, STATE_ ) values ( '1501', 'topic', 'topic:4:1003', '1401', '1401', '1401', '1401', 'a', null, null, null, null, null, 'a', 0, 'integer', '2021-03-10 11:06:20.967', null, null, null, 10, '10', null, 'CREATED' );* insert into ACT_HI_ACTINST ( ID_, PARENT_ACT_INST_ID_, PROC_DEF_KEY_, PROC_DEF_ID_, ROOT_PROC_INST_ID_, PROC_INST_ID_, EXECUTION_ID_, ACT_ID_, TASK_ID_, CALL_PROC_INST_ID_, CALL_CASE_INST_ID_, ACT_NAME_, ACT_TYPE_, ASSIGNEE_, START_TIME_, END_TIME_, DURATION_, ACT_INST_STATE_, SEQUENCE_COUNTER_, TENANT_ID_, REMOVAL_TIME_ ) values ( 'Activity_0y8yfpp:1503', '1401', 'topic', 'topic:4:1003', '1401', '1401', '1502', 'Activity_0y8yfpp', null, null, null, '外部任务2', 'serviceTask', null, '2021-03-10 11:06:20.98', null, null, 0, 5, 'a', null );* insert into ACT_RU_EXECUTION ( ID_, ROOT_PROC_INST_ID_, PROC_INST_ID_, BUSINESS_KEY_, PROC_DEF_ID_, ACT_ID_, ACT_INST_ID_, IS_ACTIVE_, IS_CONCURRENT_, IS_SCOPE_, IS_EVENT_SCOPE_, PARENT_ID_, SUPER_EXEC_, SUPER_CASE_EXEC_, CASE_INST_ID_, SUSPENSION_STATE_, CACHED_ENT_STATE_, SEQUENCE_COUNTER_, TENANT_ID_, REV_ ) values ( '1502', null, '1401', null, 'topic:4:1003', 'Activity_0y8yfpp', 'Activity_0y8yfpp:1503', true, false, true, false, '1401', null, null, null, 1, 128, 5, 'a', 1 );* insert into ACT_RU_VARIABLE ( ID_, TYPE_, NAME_, PROC_INST_ID_, EXECUTION_ID_, CASE_INST_ID_, CASE_EXECUTION_ID_, TASK_ID_, BYTEARRAY_ID_, DOUBLE_, LONG_, TEXT_, TEXT2_, VAR_SCOPE_, SEQUENCE_COUNTER_, IS_CONCURRENT_LOCAL_, TENANT_ID_, REV_ ) values ( '1501', 'integer', 'a', '1401', '1401', null, null, null, null, null, 10, '10', null, '1401', 1, false, 'a', 1 );* insert into ACT_RU_EXT_TASK ( ID_, WORKER_ID_, TOPIC_NAME_, LOCK_EXP_TIME_, RETRIES_, ERROR_MSG_, ERROR_DETAILS_ID_, SUSPENSION_STATE_, EXECUTION_ID_, PROC_INST_ID_, PROC_DEF_ID_, PROC_DEF_KEY_, ACT_ID_, ACT_INST_ID_, TENANT_ID_, PRIORITY_, REV_ ) values ( '1504', null, 'topic2', null, null, null, null, 1, '1502', '1401', 'topic:4:1003', 'topic', 'Activity_0y8yfpp', 'Activity_0y8yfpp:1503', 'a', 100, 1 );* delete FROM ACT_RU_EXT_TASK WHERE ID_ = '1406' and REV_ = 2;* update ACT_RU_EXECUTION set REV_ = 2, PROC_DEF_ID_ = 'topic:4:1003', BUSINESS_KEY_ = null, ACT_ID_ = null, ACT_INST_ID_ = '1401', IS_ACTIVE_ = false, IS_CONCURRENT_ = false, IS_SCOPE_ = true, IS_EVENT_SCOPE_ = false, PARENT_ID_ = null, SUPER_EXEC_ = null, SUSPENSION_STATE_ = 1, CACHED_ENT_STATE_ = 16, SEQUENCE_COUNTER_ = 4, TENANT_ID_ = 'a' WHERE ID_ = '1401' and REV_ = 1;* delete FROM ACT_RU_EXECUTION WHERE ID_ = '1404' and REV_ = 1;* UPDATE ACT_HI_ACTINST SET EXECUTION_ID_ = '1404', PROC_DEF_KEY_ = 'topic', PROC_DEF_ID_ = 'topic:4:1003', ACT_ID_ = 'Activity_05ut8e6', ACT_NAME_ = '外部任务1', ACT_TYPE_ = 'serviceTask', PARENT_ACT_INST_ID_ = '1401' , END_TIME_ = '2021-03-10 11:06:20.974' , DURATION_ = 84974 , ACT_INST_STATE_ = 4 WHERE ID_ = 'Activity_05ut8e6:1405'*/@Testpublic void complete2() {String externalTaskId = "1406";String workerId = "peng3";externalTaskService.complete(externalTaskId, workerId, Variables.createVariables().putValue("a", 10));}

 /*** delete from ACT_RU_EXT_TASK where ID_ = ? and REV_ = ?* delete from ACT_RU_VARIABLE where ID_ = ? and REV_ = ?* delete from ACT_RU_EXECUTION where ID_ = ? and REV_ = ?* UPDATE ACT_HI_ACTINST SET EXECUTION_ID_ = ?, PROC_DEF_KEY_ = ?, PROC_DEF_ID_ = ?, ACT_ID_ = ?, ACT_NAME_ = ?, ACT_TYPE_ = ?, PARENT_ACT_INST_ID_ = ? , END_TIME_ = ? , DURATION_ = ? , ACT_INST_STATE_ = ? WHERE ID_ = ?* update ACT_HI_PROCINST set PROC_DEF_ID_ = ?, PROC_DEF_KEY_ = ?, BUSINESS_KEY_ = ?, END_ACT_ID_ = ?, DELETE_REASON_ = ?, SUPER_PROCESS_INSTANCE_ID_ = ?, STATE_ = ? , END_TIME_ = ? , DURATION_ = ? where ID_ = ?*/@Testpublic void deleteProcessInstance() {String processInstanceId = "1901";String deleteReason = "测试删除";runtimeService.deleteProcessInstance(processInstanceId, deleteReason);}

六、重试次数优先级批更新及解锁

更新重试次数与优先级：

 @Testpublic void setRetries() {String externalTaskId = "2006";int retries = 4;externalTaskService.setRetries(externalTaskId, retries);}@Testpublic void setPriority() {String externalTaskId = "2006";int priority = 1000;externalTaskService.setPriority(externalTaskId, priority);}

批量更新重试次数：

    @Testpublic void updateRetries() {externalTaskService.updateRetries().externalTaskIds("2006", "2106").set(10);}

    @Testpublic void updateRetriesExternalTaskQuery() {ExternalTaskQuery externalTaskQuery = externalTaskService.createExternalTaskQuery();externalTaskService.updateRetries().externalTaskQuery(externalTaskQuery).set(100);}

外部任务过期时间更新：

    @Testpublic void fetchAndLock2() {//锁定时间变更为：当前时间+5分钟Date date = new Date();date.setMinutes(date.getMinutes() - 5);ClockUtil.setCurrentTime(date);String externalTaskId = "2006";String workerId = "peng3";long lockDuration = 1000 * 60 * 10;externalTaskService.extendLock(externalTaskId, workerId, lockDuration);//或不用ClockUtil，直接        long lockDuration = 1000 * 60 * 5;}

七、解锁外部任务

    @Testpublic void unlock() {String externalTaskId = "2006";externalTaskService.unlock(externalTaskId);}

八、外部任务错误处理

 /**** update ACT_RU_EXT_TASK SET REV_ = ?, WORKER_ID_ = ?, TOPIC_NAME_ = ?, LOCK_EXP_TIME_ = ?, RETRIES_ = ?, ERROR_MSG_ = ?, ERROR_DETAILS_ID_ = ?, EXECUTION_ID_ = ?, PROC_INST_ID_ = ?, PROC_DEF_ID_ = ?, PROC_DEF_KEY_ = ?, ACT_ID_ = ?, ACT_INST_ID_ = ?, SUSPENSION_STATE_ = ?, PRIORITY_ = ? where ID_= ? and REV_ = ? */@Testpublic void handleFailure() {String externalTaskId = "2006";String workerId = "peng3";String errorMsg = "处理失败，稍后重试";int retries = 4;int retryTimeout = 1000 * 60 * 2;externalTaskService.handleFailure(externalTaskId, workerId, errorMsg, retries, retryTimeout);}

添加错误详情信息：


@Testpublic void handleFailure2() {String exceptionStacktrace = "";try {int a = 1 / 0;} catch (Exception ex) {exceptionStacktrace = ExceptionUtil.getExceptionStacktrace(ex);}String externalTaskId = "2006";String workerId = "peng3";String errorMsg = "处理失败";int retries = 4;int retryTimeout = 1000 * 60 * 2;externalTaskService.handleFailure(externalTaskId, workerId, errorMsg, exceptionStacktrace, retries, retryTimeout);}@Testpublic void handleFailure3() {String externalTaskId = "2006";String externalTaskErrorDetails = externalTaskService.getExternalTaskErrorDetails(externalTaskId);System.out.println("#########");System.out.println(externalTaskErrorDetails);}

九、外部任务错误手工查询并处理

    @Testpublic void handleFailure2() {String exceptionStacktrace = "";try {int a = 1 / 0;} catch (Exception ex) {exceptionStacktrace = ExceptionUtil.getExceptionStacktrace(ex);}String externalTaskId = "2006";String workerId = "peng3";String errorMsg = "处理失败";int retries = 0;int retryTimeout = 1000 * 60 * 2;externalTaskService.handleFailure(externalTaskId, workerId, errorMsg, exceptionStacktrace, retries, retryTimeout);}

这时候无法再被抓取与锁定

此时我们可以通过手工查询处理，然后去完成

    /*** select distinct RES.* from ACT_RU_INCIDENT RES order by RES.ID_ asc LIMIT ? OFFSET ?*/@Testpublic void createIncidentQuery() {List<Incident> incidentList = runtimeService.createIncidentQuery().list();for (Incident incident : incidentList) {System.out.println("####");System.out.println(incident.getId());System.out.println("####");}}

    @Testpublic void createIncidentQueryComplete() {List<Incident> incidentList = runtimeService.createIncidentQuery().list();for (Incident incident : incidentList) {externalTaskService.complete(incident.getConfiguration(), "peng3");}}

十、外部任务与错误边界事件

 <bpmn:serviceTask id="Activity_1puff4g" name="外部任务1" camunda:type="external" camunda:topic="topic1" camunda:taskPriority="50"><bpmn:incoming>Flow_0rhhlhl</bpmn:incoming><bpmn:outgoing>Flow_196a355</bpmn:outgoing></bpmn:serviceTask>

  <bpmn:error id="Error_0k4jkdg" name="error1" />

@Testpublic void fetchAndLock() {List<LockedExternalTask> lockedExternalTaskList = externalTaskService.fetchAndLock(5, "peng3", true).topic("topic1", 1000 * 10 * 5).execute();for (LockedExternalTask lockedExternalTask : lockedExternalTaskList) {System.out.println("#########");System.out.println(lockedExternalTask.getPriority());System.out.println(lockedExternalTask.getId());System.out.println(lockedExternalTask.getTopicName());System.out.println("#########");}}@Testpublic void handleBpmnError() {String externalTaskId = "3105";String workerId = "peng3";String errorCode = "error1";externalTaskService.handleBpmnError(externalTaskId, workerId, errorCode);}

此时会走到“任务2”（如果没有抛出错误，完成外部任务时，会走到任务1）

完成任务2，实例完成

代码参考：https://github.com/zhoupengwa/ProcessEngineDemo

学习来源：腾讯课堂

camunda框架学习（十二）外部任务系列相关推荐

PyTorch框架学习十二——损失函数
PyTorch框架学习十二--损失函数一.损失函数的作用二.18种常见损失函数简述 1.L1Loss(MAE) 2.MSELoss 3.SmoothL1Loss 4.交叉熵CrossEntropy ...
PyTorch框架学习十九——模型加载与保存
PyTorch框架学习十九--模型加载与保存一.序列化与反序列化二.PyTorch中的序列化与反序列化 1.torch.save 2.torch.load 三.模型的保存 1.方法一:保存整个Mo ...
PyTorch框架学习十八——Layer Normalization、Instance Normalization、Group Normalization
PyTorch框架学习十八--Layer Normalization.Instance Normalization.Group Normalization 一.为什么要标准化? 二.BN.LN.IN. ...
PyTorch框架学习十六——正则化与Dropout
PyTorch框架学习十六--正则化与Dropout 一.泛化误差二.L2正则化与权值衰减三.正则化之Dropout 补充: 这次笔记主要关注防止模型过拟合的两种方法:正则化与Dropout. 一 ...
PyTorch框架学习十五——可视化工具TensorBoard
PyTorch框架学习十五--可视化工具TensorBoard 一.TensorBoard简介二.TensorBoard安装及测试三.TensorBoard的使用 1.add_scalar() 2 ...
PyTorch框架学习十四——学习率调整策略
PyTorch框架学习十四--学习率调整策略一._LRScheduler类二.六种常见的学习率调整策略 1.StepLR 2.MultiStepLR 3.ExponentialLR 4.Cosin ...
PyTorch框架学习十——基础网络层（卷积、转置卷积、池化、反池化、线性、激活函数）
PyTorch框架学习十--基础网络层(卷积.转置卷积.池化.反池化.线性.激活函数) 一.卷积层二.转置卷积层三.池化层 1.最大池化nn.MaxPool2d 2.平均池化nn.AvgPool2 ...
（转）SpringMVC学习(十二)——SpringMVC中的拦截器
http://blog.csdn.net/yerenyuan_pku/article/details/72567761 SpringMVC的处理器拦截器类似于Servlet开发中的过滤器Filter, ...
Spring Boot 框架学习笔记(二）（配置文件与数据注入 yaml基本语法 JSR303数据验证多环境切换）
Spring Boot 框架学习笔记(二) 六.appliaction.properties配置与数据注入 6.1 `@Value`注解测试注入数据读取输入流 6.2 读取配置文件数据注入单文件 ...
OpenCV与图像处理学习十二——图像形状特征之HOG特征
OpenCV与图像处理学习十二--图像形状特征之HOG特征一.图像特征理解 1.1 颜色特征 1.2 纹理特征 1.3 形状特征 1.4 空间关系特征二.形状特征描述 2.1 HOG特征 2.1. ...