邻接表 vs 邻接矩阵：Python实现DAG的5种性能优化技巧

在构建复杂的任务调度系统、机器学习流水线或CI/CD工作流时，有向无环图（DAG）往往是背后的核心引擎。它清晰地定义了任务间的依赖关系，确保执行顺序的正确性。然而，当任务节点数量从几十个膨胀到成千上万个，当依赖关系变得错综复杂时，一个朴素的DAG实现很快就会成为性能瓶颈。这时，开发者面临一个关键抉择：如何选择底层的数据结构？是使用内存占用更灵活的邻接表，还是查询效率可能更高的邻接矩阵？这不仅仅是理论上的选择，它直接决定了系统在高并发、大规模数据处理场景下的响应速度和资源消耗。

本文面向那些已经熟悉DAG基本概念，但在实际项目中遭遇性能挑战的中高级Python开发者。我们将深入剖析邻接表与邻接矩阵在不同操作场景下的性能表现，并通过一系列基准测试数据来量化它们的差异。更重要的是，我们将超越基础的数据结构选择，探讨五种切实可行的性能优化技巧，包括缓存策略、并行执行优化、内存布局调整等，帮助你将DAG引擎的性能提升一个数量级。无论你是在优化一个内部的ETL流程，还是构建一个面向用户的计算服务平台，这些技巧都能提供直接的参考价值。

1. 数据结构选型：邻接表与邻接矩阵的深度性能剖析

选择哪种数据结构来表示DAG，绝非凭感觉决定。邻接表和邻接矩阵在空间复杂度、时间复杂度上各有胜负，其优劣高度依赖于你的具体操作模式。简单来说，邻接表使用字典（或列表）来存储每个节点的出边列表，适合稀疏图；而邻接矩阵则使用一个二维数组（或列表的列表），用矩阵元素的值表示节点间是否存在边，更适合稠密图或需要频繁进行邻接关系查询的场景。

为了更直观地对比，我们来看一个具体的性能对照表，它基于对节点数（N）从100到10000的图进行常见操作的平均时间复杂度分析：

操作邻接表 (平均情况)邻接矩阵适用场景与说明
添加边O(1)O(1)两者都很快，但邻接表需追加到列表，矩阵是直接赋值。
删除边O(d)O(1)邻接表需要遍历邻居列表(d为出度)，矩阵直接。删除频繁选矩阵。
检查边(u->v)O(d)O(1)邻接表需要遍历u的邻居列表，矩阵是常量时间访问。查询频繁选矩阵。
遍历节点所有出边O(d)O(N)邻接表仅遍历实际邻居，矩阵需扫描整行。遍历操作多选邻接表。
空间占用O(N + E)O(N²)E为边数。对于稀疏图(E << N²)，邻接表空间优势巨大。
拓扑排序 (Kahn算法)O(N + E)O(N²)邻接表能高效利用图的稀疏性，矩阵实现会带来不必要的开销。

提示：这里的“稠密”与“稀疏”是一个相对概念。一个经验法则是，如果边数E接近节点数N的平方（即 E ≈ N²），那么图可以被认为是稠密的；如果E远小于N²，则是稀疏的。在任务调度类DAG中，依赖关系通常是局部和链式的，绝大多数属于稀疏图。

从表格可以看出，如果你的DAG操作以添加边和遍历邻接关系（如执行拓扑排序）为主，那么邻接表是更自然、更高效的选择。这也是为什么绝大多数通用图算法库默认采用邻接表的原因。然而，如果你的应用需要极高频地随机查询两个特定节点间是否存在依赖关系（例如，在动态更新的工作流中实时检查某个任务是否已被另一个任务阻塞），那么邻接矩阵的O(1)查询时间将带来显著优势，即使它牺牲了空间。

让我们用代码来感受一下这种差异。下面是一个简单的基准测试，对比在稀疏图（边数约为节点数的2倍）和半稠密图（边数约为节点数的平方的一半）下，执行10万次“检查边是否存在”操作的耗时：

import time
from collections import defaultdict
import random

class AdjacencyListDAG:
    def __init__(self, n):
        self.n = n
        self.graph = defaultdict(list)

    def add_edge(self, u, v):
        self.graph[u].append(v)

    def has_edge(self, u, v):
        return v in self.graph[u]

class AdjacencyMatrixDAG:
    def __init__(self, n):
        self.n = n
        # 使用布尔列表的列表，节省一些内存
        self.matrix = [[False] * n for _ in range(n)]

    def add_edge(self, u, v):
        self.matrix[u][v] = True

    def has_edge(self, u, v):
        return self.matrix[u][v]

def benchmark_edge_query(dag_impl, edges_to_test, iterations=100000):
    """基准测试边查询操作"""
    start = time.perf_counter()
    for _ in range(iterations):
        u, v = random.choice(edges_to_test)
        _ = dag_impl.has_edge(u, v)
    end = time.perf_counter()
    return end - start

# 测试配置
N = 500  # 500个节点
sparse_edges = [(i, (i+1) % N) for i in range(N)] + [(i, (i+2) % N) for i in range(N)]  # 约1000条边
dense_edges = [(i, j) for i in range(N) for j in range(N) if i != j and random.random() < 0.5]  # 约半稠密

print("稀疏图 (N=500, E≈1000) 边查询基准测试:")
sparse_list = AdjacencyListDAG(N)
sparse_matrix = AdjacencyMatrixDAG(N)
for u, v in sparse_edges:
    sparse_list.add_edge(u, v)
    sparse_matrix.add_edge(u, v)

time_list_sparse = benchmark_edge_query(sparse_list, sparse_edges)
time_matrix_sparse = benchmark_edge_query(sparse_matrix, sparse_edges)
print(f"  邻接表耗时: {time_list_sparse:.4f} 秒")
print(f"  邻接矩阵耗时: {time_matrix_sparse:.4f} 秒")
print(f"  矩阵比表快: {time_list_sparse/time_matrix_sparse:.2f} 倍")

运行这段代码，你可能会发现，即使在稀疏图上，对于纯粹的随机边查询，邻接矩阵也可能比邻接表快数倍甚至数十倍。这直观地展示了操作模式对性能影响的决定性作用。因此，在做选择前，务必先对你的DAG进行操作画像分析：统计在典型工作负载下，各种操作（增边、删边、查边、遍历邻居）的频率分布。

2. 超越基础：针对邻接表的5种高级优化技巧

假设经过分析，你的DAG属于典型的稀疏图，且操作以添加边和拓扑遍历为主，邻接表是更合适的基础。但这仅仅是开始。一个生产级的、高性能的DAG实现需要在邻接表的基础上进行多层次的优化。下面我们探讨五种能带来实质性提升的技巧。

2.1 技巧一：使用array或numpy数组替代列表存储邻接关系

Python的list虽然灵活，但在存储大量整数（节点ID）时，其内存开销和访问效率并非最优。每个list对象和其中的每个int对象都有额外的开销。对于节点ID连续或范围已知的DAG，可以使用array模块的'I'（无符号整型）或'i'（有符号整型）类型数组，或者直接使用numpy.ndarray。

  • array.array： 内存紧凑，操作类似列表，但元素类型固定。
  • numpy.ndarray： 向量化操作潜力大，与科学计算栈兼容性好，但引入额外依赖。

优化示例：将邻接表从defaultdict(list)改为defaultdict(lambda: array('I'))。

from array import array
from collections import defaultdict

class OptimizedAdjListDAG:
    def __init__(self):
        # 使用array('I')存储无符号整数邻居ID
        self._graph = defaultdict(lambda: array('I'))
        self._reverse_graph = defaultdict(lambda: array('I'))

    def add_edge(self, u: int, v: int):
        """假设节点ID为整数"""
        # array的append操作和list类似，但存储更紧凑
        self._graph[u].append(v)
        self._reverse_graph[v].append(u)

    def get_successors(self, u: int):
        """返回节点的后继节点数组视图，避免复制"""
        return self._graph[u]

这种优化在节点数超过数万时，能显著减少内存占用（可能减少30%-50%），并且由于数据在内存中更连续，遍历操作（如拓扑排序中访问所有邻居）的缓存命中率更高，从而提升速度。

2.2 技巧二：实现惰性入度计算与缓存

在Kahn拓扑排序算法中，我们需要反复获取每个节点的入度（即前驱节点数量）。一个常见的低效做法是在每次执行排序时，都通过len(self._reverse_graph[node])实时计算。对于静态DAG或边不频繁变动的DAG，我们可以缓存入度值。

class CachedInDegreeDAG:
    def __init__(self):
        self._graph = defaultdict(list)
        self._in_degree = defaultdict(int)  # 入度缓存

    def add_edge(self, u, v):
        self._graph[u].append(v)
        self._in_degree[v] += 1  # 添加边时更新缓存
        # 这里省略了环检测

    def remove_edge(self, u, v):
        if v in self._graph[u]:
            self._graph[u].remove(v)
            self._in_degree[v] -= 1  # 删除边时更新缓存

    def topological_sort_kahn(self):
        """使用缓存入度的Kahn算法"""
        # 初始化时直接使用缓存字典，无需再计算len
        zero_degree_nodes = [n for n in self._graph if self._in_degree[n] == 0]
        # 注意：需要包含所有节点，包括那些只有入边没有出边的“叶子”节点。
        # 更严谨的做法是维护一个所有节点的集合。
        result = []
        while zero_degree_nodes:
            node = zero_degree_nodes.pop()
            result.append(node)
            for neighbor in self._graph[node]:
                self._in_degree[neighbor] -= 1
                if self._in_degree[neighbor] == 0:
                    zero_degree_nodes.append(neighbor)
        # 排序后，如果需要恢复原始入度，需要备份或重新计算
        return result

注意：这种缓存策略在DAG结构动态变化（频繁增删边）时，需要仔细维护缓存的一致性，否则会引入难以调试的错误。对于静态DAG（构建后不再修改），此优化效果最佳。

2.3 技巧三：采用迭代DFS进行拓扑排序与环检测

递归实现的深度优先搜索（DFS）虽然简洁，但在处理超大规模图时，可能因Python递归深度限制（默认约1000层）或函数调用开销而受限。我们可以用显式的栈（list）来实现迭代DFS，用于拓扑排序（基于完成时间的逆序）和环检测。

迭代DFS环检测的核心是，除了记录节点的访问状态（未访问、访问中、已访问），我们还需要在栈中记录当前节点的下一个待访问邻居的索引，以模拟递归的回溯过程。

def has_cycle_iterative(self):
    """
    使用迭代DFS检测图中是否存在环。
    返回True如果存在环。
    """
    WHITE, GRAY, BLACK = 0, 1, 2  # 未访问，访问中，已访问
    color = {node: WHITE for node in self._graph}

    for start_node in self._graph:
        if color[start_node] != WHITE:
            continue

        # 栈元素为 (node, next_neighbor_index)
        stack = [(start_node, 0)]
        while stack:
            node, index = stack[-1]

            if color[node] == WHITE:
                color[node] = GRAY

            # 如果这个节点的所有邻居都已处理完
            if index >= len(self._graph.get(node, [])):
                color[node] = BLACK
                stack.pop()
                continue

            # 处理下一个邻居
            neighbor = self._graph[node][index]
            # 更新栈顶元素，指向下一个待处理的邻居
            stack[-1] = (node, index + 1)

            if color[neighbor] == GRAY:
                # 遇到灰色节点，说明有后向边，存在环
                return True
            if color[neighbor] == WHITE:
                stack.append((neighbor, 0))

    return False

迭代版本不仅避免了递归深度问题，而且在某些情况下由于减少了函数调用开销，性能更优。对于拓扑排序，类似的迭代DFS可以在节点被标记为BLACK时，将其加入一个结果列表，最后将列表反转即可得到拓扑序。

2.4 技巧四：利用concurrent.futures实现并行节点执行

DAG的拓扑排序给出了一个线性的、满足依赖关系的执行顺序。然而，在这个顺序中，入度同时变为0的多个节点彼此之间是没有依赖关系的，它们可以并行执行。这是DAG执行阶段最直接的性能加速点。

Python的concurrent.futures模块提供了高级的线程池（ThreadPoolExecutor）和进程池（ProcessPoolExecutor）接口，非常适合这种“任务并行”的场景。我们需要修改执行器，使其能够发现并并发执行所有当前可运行的节点（入度为0的节点集），而不是一次只执行一个。

from concurrent.futures import ThreadPoolExecutor, as_completed
from collections import deque

class ParallelDAGExecutor:
    def __init__(self, dag, max_workers=None):
        self.dag = dag
        self.max_workers = max_workers

    def execute(self):
        """并行执行DAG中的所有任务"""
        in_degree = {node: len(self.dag._reverse_graph.get(node, [])) for node in self.dag._graph}
        # 使用deque作为初始零入度节点队列
        zero_degree_queue = deque([n for n, d in in_degree.items() if d == 0])
        results = {}
        total_nodes = len(self.dag._graph)

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 未来任务映射：{future: node_id}
            future_to_node = {}

            while len(results) < total_nodes:
                # 提交所有当前可并行执行的任务
                while zero_degree_queue:
                    node = zero_degree_queue.popleft()
                    task_func, args, kwargs = self.dag._tasks.get(node, (None, (), {}))
                    if task_func:
                        # 提交任务到线程池
                        future = executor.submit(task_func, *args, **kwargs)
                        future_to_node[future] = node
                    else:
                        # 没有任务，直接标记完成
                        results[node] = None

                if not future_to_node:
                    # 没有任务在执行，也没有新任务可提交，但结果未集齐，说明有环或逻辑错误
                    if len(results) < total_nodes:
                        raise RuntimeError("DAG执行停滞，可能存在环或依赖未满足。")

                # 等待至少一个任务完成
                done, _ = as_completed(future_to_node.keys()), None
                for future in done:
                    node = future_to_node.pop(future)
                    try:
                        result = future.result()
                        results[node] = result
                        print(f"节点 {node} 执行完成，结果: {result}")
                    except Exception as exc:
                        print(f"节点 {node} 执行产生异常: {exc}")
                        # 根据策略决定：是终止所有任务，还是继续执行其他不依赖此节点的任务？
                        raise exc

                    # 节点完成后，更新其所有后继节点的入度
                    for successor in self.dag._graph.get(node, []):
                        in_degree[successor] -= 1
                        if in_degree[successor] == 0:
                            zero_degree_queue.append(successor)

        return results

这个并行执行器的核心逻辑是：

  1. 维护一个动态的零入度节点队列。
  2. 使用线程池并发执行队列中的所有节点任务。
  3. 每当一个任务完成，就更新其所有后继节点的入度，并将新产生的零入度节点加入队列。
  4. 循环直到所有节点执行完毕。

提示：选择ThreadPoolExecutor还是ProcessPoolExecutor取决于任务类型。如果任务是I/O密集型（如网络请求、文件读写），使用线程池。如果任务是CPU密集型（如大量数值计算）且任务间数据交换少，使用进程池可以绕过GIL限制，但要注意进程间通信开销。max_workers参数需要根据运行环境（CPU核心数、I/O负载）进行调优。

2.5 技巧五：应用内存视图与零拷贝优化大规模遍历

当DAG规模极大，进行全图遍历（如计算所有节点的某种指标）时，数据在内存中的移动会成为开销。如果使用numpy数组存储邻接关系，可以利用其**内存视图（memory view）和花式索引（fancy indexing）**来实现“零拷贝”或“向量化”操作，极大提升批量处理速度。

假设我们将所有节点的邻接列表存储在一个大的numpy数组中，并用另一个数组记录每个节点邻接列表的起始索引（类似于CSR压缩稀疏行格式）。这种格式在进行以节点为单位的批量操作时效率极高。

import numpy as np

class CSRFormatDAG:
    """
    使用压缩稀疏行格式存储邻接关系，优化遍历性能。
    适用于静态或很少修改的DAG。
    """
    def __init__(self, num_nodes):
        self.num_nodes = num_nodes
        self.indptr = np.zeros(num_nodes + 1, dtype=np.int32)  # 行指针
        self.indices = np.array([], dtype=np.int32)  # 列索引（邻居ID）
        # 需要从边列表构建，此处省略构建过程

    def build_from_edge_list(self, edge_list):
        """从边列表构建CSR格式（简化版）"""
        # 1. 计算每个节点的出度
        out_degree = np.zeros(self.num_nodes, dtype=np.int32)
        for u, v in edge_list:
            out_degree[u] += 1
        # 2. 计算indptr
        self.indptr[1:] = np.cumsum(out_degree)
        # 3. 填充indices
        self.indices = np.zeros(self.indptr[-1], dtype=np.int32)
        temp_pos = self.indptr.copy()
        for u, v in edge_list:
            pos = temp_pos[u]
            self.indices[pos] = v
            temp_pos[u] += 1

    def get_successors_view(self, u):
        """获取节点u的所有后继节点的内存视图，无数据拷贝"""
        start, end = self.indptr[u], self.indptr[u+1]
        return self.indices[start:end]  # 这是一个numpy数组的视图，不是副本

    def batch_process_successors(self, node_list, process_func):
        """批量处理一批节点的所有后继节点，利用向量化潜力"""
        # 例如，计算这批节点所有后继节点的平均ID
        all_successors = []
        for u in node_list:
            succ = self.get_successors_view(u)
            all_successors.append(succ)
        # 这里all_successors是视图列表，实际数据仍在self.indices中
        # 可以传递给一个能处理数组的process_func
        return process_func(all_successors)

CSR格式是科学计算和高性能图计算库（如scipy.sparse）中的标准存储格式。它将邻接矩阵的非零元素（即边）压缩存储，既节省了稀疏图的空间，又提供了高效的按行遍历能力。虽然构建CSR需要预处理成本，但对于构建后反复执行遍历和查询的DAG应用来说，这种一次性的开销是值得的。

3. 实战：优化一个机器学习特征工程流水线DAG

让我们将这些技巧应用到一个具体的场景：一个机器学习特征工程流水线。假设我们有数十个特征处理步骤（如缺失值填充、标准化、编码、特征交叉等），它们之间存在复杂的依赖关系，构成了一个DAG。我们的目标是高效地执行这个流水线，并能够方便地添加新步骤或调整依赖。

初始实现（朴素邻接表）： 我们会有一个FeatureNode类代表处理步骤，一个简单的DAG类来管理依赖和执行。执行可能是顺序的，性能瓶颈在于串行执行和频繁的邻居列表遍历。

优化步骤：

  1. 数据结构升级： 由于步骤数量固定（构建后不变），我们为每个节点分配一个整数ID。使用List[array('I')]代替defaultdict(list)来存储邻接表，通过节点ID直接索引，减少字典查找开销。
  2. 入度缓存： 在DAG构建阶段（添加所有边之后），计算并缓存每个节点的入度数组in_degree = np.zeros(n, dtype=int)。
  3. 并行执行器： 采用上述的ParallelDAGExecutor，并设置max_workers为CPU核心数，以并行执行独立的特征处理步骤。每个FeatureNode的task_func封装了具体的特征变换逻辑。
  4. 结果缓存与复用： 对于昂贵的特征计算步骤，在其节点对象中加入缓存机制。如果输入数据哈希未变，则直接返回缓存结果，避免重复计算。这需要DAG执行器能够传递节点间的数据。
  5. 向量化节点： 对于某些数值计算步骤（如标准化），利用numpy或pandas的向量化操作重写task_func，使其能一次性处理整个数据集的一列或多列，而不是循环处理每个样本。

经过这些优化，一个原本需要数分钟才能完成的大型特征流水线，可能被缩短到几十秒内完成。性能提升主要来自于：1) 并行化利用了多核；2) 紧凑的数据结构减少了内存访问延迟；3) 向量化操作提升了单任务CPU效率。

4. 性能基准测试与量化对比

理论分析固然重要，但用数据说话更有说服力。我们可以设计一个基准测试框架，来量化不同优化技巧带来的收益。测试可以围绕以下几个维度展开：

  • 构建时间： 随机生成一个包含N个节点、E条边的DAG，测量从边列表构建出完整数据结构（朴素邻接表、优化邻接表、CSR格式）所需的时间。
  • 内存占用： 使用sys.getsizeof()或pympler等工具，测量不同数据结构下，存储相同图所需的内存大小。
  • 拓扑排序速度： 对同一张图，分别用递归DFS、迭代DFS、Kahn算法（使用缓存入度）进行拓扑排序，比较耗时。
  • 并行执行加速比： 固定DAG结构和任务（如模拟计算），分别用单线程执行器和多线程/多进程执行器运行，计算加速比（单线程时间/多线程时间）。绘制加速比随工作线程数变化的曲线，观察是否接近线性加速，以及在何处达到瓶颈。
  • 边查询吞吐量： 随机生成大量边查询请求，测试邻接表（v in list）、邻接矩阵、以及基于set的邻接表（v in set）的查询吞吐量（QPS）。

下面是一个简单的拓扑排序性能对比测试框架示例：

import timeit
import random
from collections import deque, defaultdict

# 这里导入各种DAG实现类：NaiveDAG, OptimizedListDAG, IterativeDFSDAG等

def generate_random_dag(nodes, edge_probability=0.1):
    """生成一个随机的DAG边列表（确保无环）"""
    edges = []
    for i in range(nodes):
        for j in range(i+1, nodes): # 只从编号小的节点指向编号大的，确保无环
            if random.random() < edge_probability:
                edges.append((i, j))
    return edges

def benchmark_topological_sort(dag_class, edges, setup_code=''):
    """基准测试特定DAG类的拓扑排序性能"""
    def test():
        dag = dag_class(len(set([u for u,_ in edges] + [v for _,v in edges])))
        for u, v in edges:
            dag.add_edge(u, v)
        return dag.topological_sort()
    # 使用timeit进行多次测量，取平均
    timer = timeit.Timer(test, setup=setup_code)
    times = timer.repeat(repeat=5, number=10) # 重复5组，每组运行10次
    return min(times) / 10  # 返回最好的一次平均时间

# 生成测试数据
N = 2000
test_edges = generate_random_dag(N, 0.05)

print(f"基准测试：节点数={N}, 边数≈{len(test_edges)}")
print("-" * 50)
# 假设我们有这些类
# results = {}
# results['Naive'] = benchmark_topological_sort(NaiveDAG, test_edges, 'from __main__ import NaiveDAG')
# results['Optimized'] = benchmark_topological_sort(OptimizedListDAG, test_edges, 'from __main__ import OptimizedListDAG')
# ... 打印结果对比

通过这样的基准测试，你可以清晰地看到，将邻接表内部的list换成array可能带来15%的排序速度提升；使用迭代DFS替代递归DFS在处理深链状DAG时可能避免递归错误并快上10%；而启用并行执行，在任务都是CPU密集型且相互独立时，可能获得接近线程数量的线性加速比。这些具体的数字，是你向团队证明优化价值、为项目选择最合适实现方案的最有力依据。

优化永无止境。除了本文讨论的这些技巧，根据具体应用场景，你还可以探索更多方向，例如使用PyPy解释器获得即时编译优化，对关键循环使用Cython或Numba进行编译，或者直接集成像NetworkX（底层用字典）或graph-tool（底层用C++，性能极高）这样的专业图库。理解底层数据结构的性能特征，并结合实际的业务操作模式进行有针对性的优化，才是打造高性能DAG系统的关键。我在重构一个旧有的任务调度系统时，正是通过将存储从默认字典列表改为整数数组，并将核心的拓扑排序算法重写为迭代版本，使得处理万级节点图的延迟从秒级降低到了毫秒级，这种提升对于用户体验来说是质的飞跃。记住，没有银弹，最好的优化永远是基于测量的优化。

# 性能优化
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

程序员学 AI（八）：AI Agent 到底是什么？从回答问题到完成任务

avatar AI Agent技术社区
cover

22道AI Agent 工程师必会的面试题，别啥都不会就出去找工作

avatar AI Agent技术社区
cover

基于AI agent的智能图书馆

avatar AI Agent技术社区