鞍钢多次重试网络异常请求超时的，自动驳回审批，做好异常捕获日志
This commit is contained in:
parent
3ce4ef018e
commit
e17c8172a4
@ -1,17 +1,23 @@
|
||||
package com.wflow.exception;
|
||||
|
||||
import lombok.Getter;
|
||||
import lombok.Data;
|
||||
|
||||
/**
|
||||
* @author : willian fu
|
||||
* @date : 2022/6/27
|
||||
*/
|
||||
@Getter
|
||||
@Data
|
||||
public class BusinessException extends RuntimeException {
|
||||
private Integer code;
|
||||
|
||||
public BusinessException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
public BusinessException(Integer code, String message) {
|
||||
super(message);
|
||||
this.code = code;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
@ -1,7 +1,6 @@
|
||||
package com.wflow.workflow.config.listener;
|
||||
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
import com.wflow.exception.BusinessException;
|
||||
import com.wflow.workflow.bean.dto.NotifyDto;
|
||||
@ -10,7 +9,6 @@ import com.wflow.workflow.service.NotifyService;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.flowable.bpmn.model.EndEvent;
|
||||
import org.flowable.common.engine.api.delegate.event.FlowableEngineEntityEvent;
|
||||
import org.flowable.engine.HistoryService;
|
||||
import org.flowable.engine.RuntimeService;
|
||||
import org.flowable.engine.delegate.event.*;
|
||||
import org.flowable.engine.runtime.ProcessInstance;
|
||||
@ -88,7 +86,7 @@ public class GlobalTaskListener extends AbstractFlowableEngineEventListener {
|
||||
if (pass != null) {
|
||||
JSONObject obj = JSONObject.parseObject(pass.toString());
|
||||
if (obj.getJSONObject("data").getInteger("code") != 200) {
|
||||
throw new BusinessException(obj.getJSONObject("data").getString("message"));
|
||||
throw new BusinessException(obj.getJSONObject("data").getInteger("code"), obj.getJSONObject("data").getString("message"));
|
||||
}
|
||||
}
|
||||
ProcessInstance instance = runtimeService.createProcessInstanceQuery()
|
||||
@ -111,11 +109,11 @@ public class GlobalTaskListener extends AbstractFlowableEngineEventListener {
|
||||
@Override
|
||||
protected void processCompletedWithTerminateEnd(FlowableEngineEntityEvent event) {
|
||||
//通过判断流程实例的endActivityId = refuse-end / cancel-end 判断是撤销还是驳回
|
||||
if (event instanceof FlowableProcessTerminatedEvent){
|
||||
if (event instanceof FlowableProcessTerminatedEvent) {
|
||||
Object cause = ((FlowableProcessTerminatedEvent) event).getCause();
|
||||
if (cause instanceof EndEvent){
|
||||
if (cause instanceof EndEvent) {
|
||||
String endNode = ((EndEvent) cause).getId();
|
||||
if ("refuse-end".equals(endNode)){
|
||||
if ("refuse-end".equals(endNode)) {
|
||||
log.debug("监听到流程[{}]被驳回", event.getProcessInstanceId());
|
||||
listenerExecutor.doProcessChangeHandler("refuse", event.getProcessInstanceId(), event.getProcessDefinitionId());
|
||||
} else if ("cancel-end".equals(endNode)) {
|
||||
@ -127,7 +125,7 @@ public class GlobalTaskListener extends AbstractFlowableEngineEventListener {
|
||||
super.processCompletedWithTerminateEnd(event);
|
||||
}
|
||||
|
||||
private void processLeaveNodeEventHandler(FlowableActivityEvent event){
|
||||
private void processLeaveNodeEventHandler(FlowableActivityEvent event) {
|
||||
log.info("流程[{}]离开ID[{}]的[{}]节点", event.getProcessInstanceId(), event.getActivityId(), event.getActivityName());
|
||||
listenerExecutor.doProcessNodeChangeHandler("leave", event.getProcessInstanceId(),
|
||||
event.getProcessDefinitionId(), event.getActivityId(), event.getActivityType());
|
||||
|
||||
@ -2,28 +2,24 @@ package com.wflow.workflow.execute;
|
||||
|
||||
import cn.hutool.core.bean.BeanUtil;
|
||||
import cn.hutool.core.collection.CollectionUtil;
|
||||
import cn.hutool.core.util.ObjectUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import cn.hutool.http.HttpException;
|
||||
import com.alibaba.fastjson2.JSONArray;
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.wflow.bean.entity.WflowModelHistorys;
|
||||
import com.wflow.mapper.WflowModelHistorysMapper;
|
||||
import com.wflow.workflow.UELTools;
|
||||
import com.wflow.workflow.bean.dto.ProcessInstanceOwnerDto;
|
||||
import com.wflow.workflow.bean.process.HttpDefinition;
|
||||
import com.wflow.workflow.bean.process.props.ApprovalProps;
|
||||
import com.wflow.workflow.bean.process.props.RootProps;
|
||||
import com.wflow.workflow.config.WflowGlobalVarDef;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.flowable.engine.RuntimeService;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.HashSet;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
|
||||
/**
|
||||
@ -43,43 +39,60 @@ public class ListenerExecutor {
|
||||
@Autowired
|
||||
private ExecutorService executorService;
|
||||
|
||||
public Object doProcessChangeHandler(String event, String instanceId, String defId){
|
||||
public Object doProcessChangeHandler(String event, String instanceId, String defId) {
|
||||
try {
|
||||
final Object[] result = {null};
|
||||
WflowModelHistorys selected = modelHistorysMapper.selectOne(
|
||||
new LambdaQueryWrapper<WflowModelHistorys>()
|
||||
.select(WflowModelHistorys::getProcessConfig)
|
||||
.eq(WflowModelHistorys::getProcessDefId, defId));
|
||||
if (Objects.nonNull(selected) && StrUtil.isNotBlank(selected.getProcessConfig())){
|
||||
if (Objects.nonNull(selected) && StrUtil.isNotBlank(selected.getProcessConfig())) {
|
||||
Map<String, Object> contextVar = uelTools.getContextVar(instanceId, defId);
|
||||
JSONObject parsed = JSONObject.parseObject(selected.getProcessConfig());
|
||||
JSONObject listener = parsed.getJSONObject("listener");
|
||||
if (Objects.nonNull(listener)){
|
||||
if (Objects.nonNull(listener)) {
|
||||
listener.getJSONArray(event).toJavaList(JSONObject.class).forEach(object -> {
|
||||
String actionType = object.getString("actionType");
|
||||
switch (actionType){
|
||||
switch (actionType) {
|
||||
case "JS":
|
||||
new JsExecute(executorService).executeVoid("action", "function action(ctx){"+ object.getString("js") +"}" , contextVar);
|
||||
new JsExecute(executorService).executeVoid("action", "function action(ctx){" + object.getString("js") + "}", contextVar);
|
||||
break;
|
||||
case "JAVA":
|
||||
new ElExecute().execute(object.getString("java"), contextVar, Object.class);
|
||||
break;
|
||||
case "HTTP":
|
||||
for (int i = 0; i < 3; i++) {
|
||||
try {
|
||||
result[0] = new HttpExecute().execute(object.getObject("http", HttpDefinition.class), executorService, Object.class, contextVar);
|
||||
break;
|
||||
} catch (HttpException e) {
|
||||
if (i == 2) {
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
return result[0];
|
||||
} catch (HttpException e) {
|
||||
log.warn("流程实例[{}]的[{}]事件触发失败：{}", instanceId, event, e.getMessage());
|
||||
JSONObject jo = new JSONObject();
|
||||
JSONObject jo1 = new JSONObject();
|
||||
jo1.put("code", 5001);
|
||||
jo1.put("message", "网络超时");
|
||||
jo.put("data", jo1);
|
||||
return jo;
|
||||
} catch (Exception e) {
|
||||
log.warn("流程实例[{}]的[{}]事件触发失败：{}", instanceId, event, e.getMessage());
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
public void doProcessNodeChangeHandler(String event, String instanceId, String defId, String nodeId, String actType){
|
||||
if (!"userTask".equals(actType)){
|
||||
public void doProcessNodeChangeHandler(String event, String instanceId, String defId, String nodeId, String actType) {
|
||||
if (!"userTask".equals(actType)) {
|
||||
return; //过滤其他非必要节点
|
||||
}
|
||||
try {
|
||||
@ -87,17 +100,17 @@ public class ListenerExecutor {
|
||||
Map map = (Map) contextVar.get(WflowGlobalVarDef.WFLOW_NODE_PROPS);
|
||||
Object nodeProps = map.get(nodeId);
|
||||
Map<String, JSONArray> nodeLis = null;
|
||||
if (nodeProps instanceof ApprovalProps){
|
||||
if (nodeProps instanceof ApprovalProps) {
|
||||
nodeLis = ((ApprovalProps) nodeProps).getListeners();
|
||||
} else if (nodeProps instanceof RootProps) {
|
||||
nodeLis = ((RootProps) nodeProps).getListeners();
|
||||
}
|
||||
if (Objects.nonNull(nodeLis) && CollectionUtil.isNotEmpty(nodeLis)){
|
||||
if (Objects.nonNull(nodeLis) && CollectionUtil.isNotEmpty(nodeLis)) {
|
||||
nodeLis.get(event).toJavaList(JSONObject.class).forEach(object -> {
|
||||
String actionType = object.getString("actionType");
|
||||
switch (actionType){
|
||||
switch (actionType) {
|
||||
case "JS":
|
||||
new JsExecute(executorService).executeVoid("action","function action(ctx){"+ object.get("js") +"}", contextVar);
|
||||
new JsExecute(executorService).executeVoid("action", "function action(ctx){" + object.get("js") + "}", contextVar);
|
||||
break;
|
||||
case "JAVA":
|
||||
new ElExecute().execute(object.getString("java"), contextVar, Object.class);
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
package com.wflow.workflow.service.impl;
|
||||
import java.util.Date;
|
||||
|
||||
import cn.hutool.cache.CacheUtil;
|
||||
import cn.hutool.cache.impl.TimedCache;
|
||||
@ -7,6 +8,8 @@ import cn.hutool.core.collection.ConcurrentHashSet;
|
||||
import cn.hutool.core.date.DateUtil;
|
||||
import cn.hutool.core.util.ObjectUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import cn.hutool.http.HttpRequest;
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
|
||||
@ -26,8 +29,10 @@ import com.wflow.workflow.bean.process.props.CcProps;
|
||||
import com.wflow.workflow.bean.vo.*;
|
||||
import com.wflow.workflow.config.WflowGlobalVarDef;
|
||||
import com.wflow.workflow.extension.cmd.RecallToHisApprovalNodeCmd;
|
||||
import com.wflow.workflow.service.*;
|
||||
import com.wflow.workflow.service.FormService;
|
||||
import com.wflow.workflow.service.BusinessDataStorageService;
|
||||
import com.wflow.workflow.service.NotifyService;
|
||||
import com.wflow.workflow.service.ProcessTaskService;
|
||||
import com.wflow.workflow.service.UserDeptOrLeaderService;
|
||||
import com.wflow.workflow.utils.FlowableUtils;
|
||||
import com.zhgd.xmgl.tenant.TenantContextHolder;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@ -38,24 +43,21 @@ import org.flowable.common.engine.impl.identity.Authentication;
|
||||
import org.flowable.engine.*;
|
||||
import org.flowable.engine.delegate.DelegateExecution;
|
||||
import org.flowable.engine.history.HistoricProcessInstance;
|
||||
import org.flowable.engine.history.HistoricProcessInstanceQuery;
|
||||
import org.flowable.engine.impl.persistence.entity.ExecutionEntity;
|
||||
import org.flowable.engine.impl.util.ExecutionGraphUtil;
|
||||
import org.flowable.engine.repository.ProcessDefinition;
|
||||
import org.flowable.engine.runtime.ActivityInstance;
|
||||
import org.flowable.engine.runtime.Execution;
|
||||
import org.flowable.engine.runtime.ProcessInstance;
|
||||
import org.flowable.engine.runtime.ProcessInstanceQuery;
|
||||
import org.flowable.task.api.Task;
|
||||
import org.flowable.task.api.TaskInfo;
|
||||
import org.flowable.task.api.TaskQuery;
|
||||
import org.flowable.task.api.history.HistoricTaskInstance;
|
||||
import org.flowable.task.service.history.NativeHistoricTaskInstanceQuery;
|
||||
import org.flowable.variable.api.history.HistoricVariableInstance;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.transaction.interceptor.TransactionAspectSupport;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.stream.Collectors;
|
||||
@ -67,43 +69,8 @@ import java.util.stream.Collectors;
|
||||
@Slf4j
|
||||
@Service("processTaskService")
|
||||
public class ProcessTaskServiceImpl implements ProcessTaskService {
|
||||
|
||||
@Autowired
|
||||
private TaskService taskService;
|
||||
|
||||
@Autowired
|
||||
private RuntimeService runtimeService;
|
||||
|
||||
@Autowired
|
||||
private NotifyService notifyService;
|
||||
|
||||
@Autowired
|
||||
private BusinessDataStorageService businessDataService;
|
||||
|
||||
@Autowired
|
||||
private UserDeptOrLeaderService userDeptOrLeaderService;
|
||||
|
||||
@Autowired
|
||||
private OrgRepositoryService orgRepositoryService;
|
||||
|
||||
@Autowired
|
||||
private HistoryService historyService;
|
||||
|
||||
@Autowired
|
||||
private RepositoryService repositoryService;
|
||||
|
||||
@Autowired
|
||||
private ManagementService managementService;
|
||||
|
||||
@Autowired
|
||||
private WflowModelHistorysMapper historysMapper;
|
||||
|
||||
@Autowired
|
||||
private WflowModelsMapper wflowModelsMapper;
|
||||
|
||||
//超时缓存，数据缓存20秒，用来存储审批人防止flowable高频调用
|
||||
private static final TimedCache<String, List<String>> taskCache = CacheUtil.newTimedCache(20000);
|
||||
|
||||
//用来存储正在处理的节点，防止并发处理
|
||||
private static final Set<String> HANDLER_NODE_LOCK = new ConcurrentHashSet<>();
|
||||
|
||||
@ -111,6 +78,31 @@ public class ProcessTaskServiceImpl implements ProcessTaskService {
|
||||
taskCache.schedulePrune(10000);
|
||||
}
|
||||
|
||||
@Value("${server.port:}")
|
||||
private String serverPort;
|
||||
@Autowired
|
||||
private TaskService taskService;
|
||||
@Autowired
|
||||
private RuntimeService runtimeService;
|
||||
@Autowired
|
||||
private NotifyService notifyService;
|
||||
@Autowired
|
||||
private BusinessDataStorageService businessDataService;
|
||||
@Autowired
|
||||
private UserDeptOrLeaderService userDeptOrLeaderService;
|
||||
@Autowired
|
||||
private OrgRepositoryService orgRepositoryService;
|
||||
@Autowired
|
||||
private HistoryService historyService;
|
||||
@Autowired
|
||||
private RepositoryService repositoryService;
|
||||
@Autowired
|
||||
private ManagementService managementService;
|
||||
@Autowired
|
||||
private WflowModelHistorysMapper historysMapper;
|
||||
@Autowired
|
||||
private WflowModelsMapper wflowModelsMapper;
|
||||
|
||||
@Override
|
||||
public Page<ProcessTaskVo> getUserTodoList(Integer pageSize, Integer pageNo, String code, String[] startTimes, String startUser, String key) {
|
||||
String userId = UserUtil.getLoginUserId();
|
||||
@ -357,7 +349,7 @@ public class ProcessTaskServiceImpl implements ProcessTaskService {
|
||||
if (ProcessHandlerParamsVo.Action.cancel.equals(params.getAction())) {
|
||||
params.getComment().setText("撤销：" + params.getComment().getText());
|
||||
}
|
||||
if (!(isComment || putComment)){
|
||||
if (!(isComment || putComment)) {
|
||||
throw new BusinessException("任务不存在，请刷新数据");
|
||||
}
|
||||
taskService.addComment(params.getTaskId(), params.getInstanceId(), JSONObject.toJSONString(params.getComment()));
|
||||
@ -420,11 +412,11 @@ public class ProcessTaskServiceImpl implements ProcessTaskService {
|
||||
//throw new BusinessException("暂不支持单独评论");
|
||||
break;
|
||||
}
|
||||
}catch (Exception e) {
|
||||
} catch (Exception e) {
|
||||
log.error("处理任务异常", e);
|
||||
// TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
|
||||
throw new BusinessException(e.getMessage());
|
||||
}finally {
|
||||
} finally {
|
||||
//移除锁
|
||||
HANDLER_NODE_LOCK.remove(lockKey);
|
||||
}
|
||||
@ -650,7 +642,29 @@ public class ProcessTaskServiceImpl implements ProcessTaskService {
|
||||
if (StrUtil.isNotBlank(params.getSignature())) {
|
||||
var.put("sign_" + task.getId(), params.getSignature());
|
||||
}
|
||||
try {
|
||||
taskService.complete(params.getTaskId(), var);
|
||||
} catch (BusinessException e) {
|
||||
if (Objects.equals(e.getCode(), 5001)) {
|
||||
log.error("", e);
|
||||
String url = "http://127.0.0.1:" + serverPort + "/xmgl/flowExceptionLog/flow/add";
|
||||
JSONObject jo = new JSONObject();
|
||||
jo.put("instanceId", task.getProcessInstanceId());
|
||||
String body = jo.toJSONString();
|
||||
log.info("工作流提交网络超时插入日志的url：{}，body：{}",url,body);
|
||||
String result2 = HttpRequest.post(url)
|
||||
.body(body)//表单内容
|
||||
.timeout(20000)//超时，毫秒
|
||||
.execute().body();
|
||||
log.info("工作流提交网络超时插入日志的返回结果：{}", result2);
|
||||
//网络超时自动驳回
|
||||
params.setAction(ProcessHandlerParamsVo.Action.refuse);
|
||||
var.put("approve_" + task.getId(), params.getAction());
|
||||
taskService.complete(params.getTaskId(), var);
|
||||
} else {
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void doRecallTask(Task task, String userId, ProcessHandlerParamsVo params) {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user