wangliwen
/
share-platform
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

异步处理待办任务内详情，大幅减少计算时间

This commit is contained in:
wangliwen 2022-12-10 16:45:56 +08:00
parent 14bbde1103
commit 72339c1f9c
1 changed files with 14 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
renren-admin/src/main/java/io/renren/modules/activiti/service/ActTaskService.java
View File

@ -65,10 +65,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import java.util.*; import java.util.*;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors; import java.util.stream.Collectors;
/** /**
@ -82,7 +79,7 @@ public class ActTaskService extends BaseServiceImpl {
private static final ObjectMapper oMapper = new ObjectMapper(); private static final ObjectMapper oMapper = new ObjectMapper();
private static final Integer CPUNUM = Runtime.getRuntime().availableProcessors(); private static final Integer CPUNUM = Runtime.getRuntime().availableProcessors();
private static final ExecutorService executor = Executors.newFixedThreadPool(CPUNUM); private static final ExecutorService executor = Executors.newWorkStealingPool(CPUNUM * 2);
public static String Task_HANDLE_STATE = "handleState"; //任务节点处理状态key public static String Task_HANDLE_STATE = "handleState"; //任务节点处理状态key
public static String Task_HANDLE_STATE_AGREE = "agree"; //同意 public static String Task_HANDLE_STATE_AGREE = "agree"; //同意
@ -193,11 +190,21 @@ public class ActTaskService extends BaseServiceImpl {
} }
List<Task> tasks = nativeTaskQuery.listPage((curPage - 1) * limit, limit); List<Task> tasks = nativeTaskQuery.listPage((curPage - 1) * limit, limit);
long count = nativeCountTaskQuery.count(); long count = nativeCountTaskQuery.count();
List<CompletableFuture> completableFutures = new ArrayList<CompletableFuture>();
List<Task> list_ = tasks.stream().map(index -> { List<Task> list_ = tasks.stream().map(index -> {
TaskQuery taskQuery = taskService.createTaskQuery(); TaskQuery taskQuery = taskService.createTaskQuery();
Task temp = taskQuery.taskId(index.getId()).includeProcessVariables().singleResult(); CompletableFuture tempTask = CompletableFuture.supplyAsync(() -> taskQuery.taskId(index.getId()).includeProcessVariables().singleResult(), executor);
Task temp = null;
try {
temp = (Task) tempTask.get();
} catch (InterruptedException e) {
logger.error("异步获取失败", e);
} catch (ExecutionException e) {
logger.error("异步获取失败", e);
}
return temp; return temp;
}).collect(Collectors.toList()); }).filter(Objects::nonNull).collect(Collectors.toList());
CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()])).join();
return new PageData<>(tasks2TaskDtos(list_), (int) count); return new PageData<>(tasks2TaskDtos(list_), (int) count);
// 以下注释为原实现方案 // 以下注释为原实现方案
// TaskQuery taskQuery = taskService.createTaskQuery(); // TaskQuery taskQuery = taskService.createTaskQuery();