3. A星算法核心数据结构

A星算法的搜索效率与路径质量，高度依赖其核心数据结构的实现。本章深入剖析开放列表（Open List）、关闭列表（Closed List）、节点状态管理以及优先队列的实现细节，为后续集成运费成本模型奠定数据结构基础。

3.1 数据结构总览与职责

下表对比了A星算法中三个核心数据结构的职责与操作特性：

3.2 开放列表（Open List）与优先队列实现

开放列表的核心需求是：始终能够快速取出当前 f(n) 值最小的节点。二叉堆（Binary Heap）是实现优先队列的标准选择。

3.2.1 二叉堆实现

Python 标准库 heapq 提供了最小堆实现，是生产环境的首选。堆中存储元组 (f, g, h, node_id, parent_id)，其中 f 值作为排序主键。

import heapq

class OpenList:
    """基于最小二叉堆的开放列表"""
    def __init__(self):
        self._heap = []
        self._entry_finder = {}  # node_id -> (f, g, h, node_id, parent_id)
        self._counter = 0        # 用于打破平局

    def push(self, node_id, f, g, h, parent_id):
        """将节点加入开放列表"""
        entry = (f, g, h, node_id, parent_id)
        self._entry_finder[node_id] = entry
        heapq.heappush(self._heap, entry)

    def pop(self):
        """取出 f 值最小的节点"""
        while self._heap:
            f, g, h, node_id, parent_id = heapq.heappop(self._heap)
            if node_id in self._entry_finder:
                del self._entry_finder[node_id]
                return node_id, f, g, h, parent_id
        return None

    def update(self, node_id, f, g, h, parent_id):
        """更新节点 f 值（惰性删除 + 重新插入）"""
        if node_id in self._entry_finder:
            del self._entry_finder[node_id]
        self.push(node_id, f, g, h, parent_id)

    def is_empty(self):
        return len(self._entry_finder) == 0

3.2.2 优先队列操作复杂度对比

3.3 关闭列表（Closed List）与快速查询

关闭列表用于标记已完全探索的节点，防止算法陷入循环或重复评估。在网格地图或路网中，节点 ID 通常是整数索引，使用 布尔数组 或 集合（Set） 实现 O(1) 查询。

class ClosedList:
    """基于集合的关闭列表"""
    def __init__(self, capacity=10000):
        self._closed = set()
        # 可选：预分配布尔数组（节点 ID 连续时更高效）
        self._bool_array = None

    def add(self, node_id):
        self._closed.add(node_id)

    def contains(self, node_id):
        return node_id in self._closed

    def clear(self):
        self._closed.clear()

# 当节点 ID 为连续整数时，布尔数组版本：
class ClosedListArray:
    def __init__(self, size):
        self._closed = [False] * size

    def add(self, node_id):
        self._closed[node_id] = True

    def contains(self, node_id):
        return self._closed[node_id]

3.4 节点状态管理

每个节点在搜索过程中需要维护以下状态信息：

• g(n)：从起点到当前节点的实际代价
• h(n)：从当前节点到终点的启发式估计代价
• f(n)：总代价 = g(n) + h(n)
• parent_id：路径中前驱节点的 ID，用于路径回溯
• status：节点状态（未访问 / 在开放列表 / 已关闭）

class NodeState:
    """节点状态管理"""
    UNVISITED = 0
    IN_OPEN = 1
    IN_CLOSED = 2

    def __init__(self, node_count):
        self.g = [float('inf')] * node_count
        self.h = [0.0] * node_count
        self.f = [float('inf')] * node_count
        self.parent = [-1] * node_count
        self.status = [NodeState.UNVISITED] * node_count

    def reset(self, start_id, goal_id):
        """重置搜索状态（保留启发式值）"""
        for i in range(len(self.g)):
            self.g[i] = float('inf')
            self.f[i] = float('inf')
            self.parent[i] = -1
            self.status[i] = NodeState.UNVISITED
        self.g[start_id] = 0.0
        self.f[start_id] = self.h[start_id]
        self.status[start_id] = NodeState.IN_OPEN

    def update_node(self, node_id, new_g, new_f, parent_id):
        """更新节点状态"""
        self.g[node_id] = new_g
        self.f[node_id] = new_f
        self.parent[node_id] = parent_id
        self.status[node_id] = NodeState.IN_OPEN

3.5 核心数据结构交互流程

下图展示了A星算法主循环中，三个数据结构之间的协作关系：

3.6 数据结构集成示例

以下代码展示了三个数据结构在A星主循环中的集成使用：

def a_star_search(graph, start_id, goal_id, heuristic_func):
    """A星算法主函数（集成核心数据结构）"""
    node_count = graph.get_node_count()
    state = NodeState(node_count)
    open_list = OpenList()
    closed_list = ClosedListArray(node_count)

    # 初始化起点
    state.h[start_id] = heuristic_func(start_id, goal_id)
    state.g[start_id] = 0.0
    state.f[start_id] = state.h[start_id]
    state.status[start_id] = NodeState.IN_OPEN
    open_list.push(start_id, state.f[start_id], state.g[start_id],
                   state.h[start_id], -1)

    while not open_list.is_empty():
        current_id, f, g, h, parent_id = open_list.pop()

        # 跳过已关闭的节点（惰性删除处理）
        if closed_list.contains(current_id):
            continue

        # 到达终点
        if current_id == goal_id:
            return reconstruct_path(state, current_id)

        # 加入关闭列表
        closed_list.add(current_id)
        state.status[current_id] = NodeState.IN_CLOSED

        # 遍历邻居
        for neighbor_id, edge_cost in graph.get_neighbors(current_id):
            if closed_list.contains(neighbor_id):
                continue

            tentative_g = state.g[current_id] + edge_cost
            if tentative_g < state.g[neighbor_id]:
                # 发现更优路径，更新节点状态
                state.g[neighbor_id] = tentative_g
                state.h[neighbor_id] = heuristic_func(neighbor_id, goal_id)
                state.f[neighbor_id] = tentative_g + state.h[neighbor_id]
                state.parent[neighbor_id] = current_id
                state.status[neighbor_id] = NodeState.IN_OPEN
                open_list.update(neighbor_id, state.f[neighbor_id],
                                 state.g[neighbor_id], state.h[neighbor_id],
                                 current_id)

    return None  # 无路径

3.7 数据结构选择决策树

根据物流场景的规模与特性，选择合适的数据结构组合：

3.8 常见陷阱与工程建议