基于Kahn算法｜动态线程池，支持扩展点并发执行｜召回｜过滤

背景

在《分布式领域扩展点设计稿》一文中，我们提到针对业务横向扩展点和纵向扩展点的编排能力。

那有这样的一种场景：针对于一次会话，同时会调很多外部服务，同时这些RPC服务会有多种直接或间接的关系，是否有更高效的方式能够让我们的一次会话时间变得更高效，同时也能够保证系统的相对稳定性呢？

设计思想

如果开发者在设计之初，对于领域边界以及子域能力划分比较清晰，那我们编排的业务扩展点就不会杂乱无章。

在SpringBoot启动时，伴随着IOC容器的初始化，领域扩展点容器也随之完成全量向量视图初始化、向量深度视图初始化、拓扑排序之后的容器初始化。

基于动态线程池思想，可以通过扩展点容器中的编排任务，动态调整线程池？根据拓扑排序后的结果，异步执行编排的任务，完成召回、过滤以及核心业务逻辑。

设计图

详细设计

1. 容器初始化时，从云端配置中心拉取扩展点编排配置文件，本地编排配置文件可以做兜底，也可以调节两者权重
2. 拿到编排数据源之后，渲染本地向量广度视图以及向量深度视图
3. 通过Kahn算法，将两个视图清洗成拓扑排序之后的容器
4. 通过权衡算法，初始化动态线程池，并预热核心线程
5. 将上述操作数据上传到我们的扩展点监控中心
6. 通过监控水位线，可动态配置线程池相关核心参数
7. 当一次会话开始时，会通过拓扑排序，并发去执行扩展点任务
8. 通过向量视图，可以针对所有扩展点做召回和过滤处理

核心算法伪代码

/*** Kahn算法** @author issavior*/
public class KahnTopologicalSort {/*** 向量广度视图*/private final Map<Integer, List<Integer>> adjList = new HashMap<>();/*** 向量深度视图*/private final Map<Integer, Integer> inDegree = new HashMap<>();/*** 拓扑排序结果*/private final List<Integer> topoOrder = new ArrayList<>();/*** 渲染向量视图和向量深度** @param u 向量头* @param v 向量尾*/public void addEdge(int u, int v) {// 渲染视图adjList.putIfAbsent(u, new ArrayList<>());adjList.get(u).add(v);// 更新入度  inDegree.put(v, inDegree.getOrDefault(v, 0) + 1);inDegree.putIfAbsent(u, 0);}/*** 拓扑排序** @return 业务节点顺序*/public List<Integer> topologicalSort() {Queue<Integer> queue = new LinkedList<>();// 将所有入度为0的节点加入队列  for (Map.Entry<Integer, Integer> entry : inDegree.entrySet()) {if (entry.getValue() == 0) {queue.offer(entry.getKey());}}while (!queue.isEmpty()) {int current = queue.poll();topoOrder.add(current);// 遍历当前节点的所有邻接点  for (int neighbor : adjList.getOrDefault(current, Collections.emptyList())) {// 减少邻接点的入度  int newInDegree = inDegree.get(neighbor) - 1;inDegree.put(neighbor, newInDegree);// 如果邻接点的入度变为0，则加入队列  if (newInDegree == 0) {queue.offer(neighbor);}}}// 检查是否所有节点都被访问过，若为有环图，初始化报错if (topoOrder.size() != inDegree.size()) {throw new IllegalStateException("Graph has a cycle and cannot be topologically sorted.");}return topoOrder;}
}