你的核心问题在于子进程无法实时同步added列表的状态，导致重复生成违反约束的路径。下面是几个针对性的解决方案，从简单适配到高效规模化，覆盖不同场景需求：

方案一：进程安全的共享集合+锁（快速适配现有代码） #

这个方案最小改动现有逻辑，把本地added列表换成跨进程安全的共享集合，同时用锁保证检查和添加操作的原子性，彻底避免并发冲突。

实现思路 #

1. 使用multiprocessing.Manager创建一个跨进程共享的集合（集合比列表更适合快速判断元素是否存在）。
2. 给共享集合配一把进程锁，确保每次只有一个子进程能执行"检查镜像串是否已存在→添加新串"的操作。
3. 子进程生成路径后，先通过锁保护的共享集合判断是否违反约束，再执行后续保存逻辑。

代码示例 #

import multiprocessing as mp
import networkx as nx

def worker(target_str, shared_added, lock, G):
    # 生成当前串的镜像串
    mirror_str = target_str[::-1]
    
    # 锁保护下执行状态检查与更新
    with lock:
        if target_str in shared_added or mirror_str in shared_added:
            return
        # 替换成你实际的节点映射和路径匹配逻辑
        start_node = sum(int(c) for c in target_str) % 2  # 示例：串的和的奇偶性为起点
        end_node = 1 - start_node  # 示例：终点为起点的对立奇偶
        paths = list(nx.shortest_simple_paths(G, start_node, end_node))
        
        if len(paths) != 1 or not path_matches_string(paths[0], target_str):
            return
        
        # 符合约束，添加到共享集合
        shared_added.add(target_str)
        # 执行后续保存/处理逻辑
        save_path(paths[0], target_str)

def path_matches_string(path, target_str):
    # 替换成你实际的路径-串匹配规则：偶数节点→0，奇数节点→1
    return ''.join(['1' if node % 2 else '0' for node in path[1:]]) == target_str

def save_path(path, target_str):
    # 替换成你的存储逻辑（文件/数据库等）
    print(f"保留路径 {path} 对应串 {target_str}")

def main():
    # 1. 构建你的无向图（这里省略具体构建逻辑）
    G = nx.Graph()
    # 示例：添加节点和边，保证奇偶节点间有唯一最短路径
    G.add_edges_from([(0,2), (2,1), (0,3), (3,1)])
    
    # 2. 生成所有候选二进制串（以3位为例）
    all_strings = [format(i, '03b') for i in range(8)]
    
    # 3. 初始化跨进程共享资源
    with mp.Manager() as manager:
        shared_added = manager.set()
        lock = manager.Lock()
        
        # 4. 创建进程池并分配任务
        with mp.Pool(processes=mp.cpu_count()) as pool:
            pool.starmap(worker, [(s, shared_added, lock, G) for s in all_strings])

if __name__ == "__main__":
    main()

适用场景 #

适合中等规模的二进制串场景，代码改动极小，能快速适配现有逻辑。缺点是锁竞争会随着进程数增加而降低效率，大规模场景下不推荐。

方案二：任务预划分（完全避免进程间冲突，最高效） #

这个方案从根源上消除共享状态的需求：预先把所有二进制串按镜像对分组，每个子进程只处理一组中的一个串，完全独立工作，不需要任何进程间通信。

实现思路 #

1. 离线预处理所有二进制串，把互为镜像的串归为一组，每组只保留一个串作为任务（比如保留字典序更小的那个）。
2. 把这些去重后的任务平均分配给各个子进程，每个子进程维护自己的本地added集合（因为任务无重叠，不会出现重复处理镜像串的情况）。
3. 子进程独立完成路径生成和约束检查，结果直接汇总即可。

代码示例 #

import multiprocessing as mp
import networkx as nx

def worker(task_group, G):
    local_added = set()
    for target_str in task_group:
        mirror_str = target_str[::-1]
        if target_str in local_added or mirror_str in local_added:
            continue
        
        # 替换成你实际的节点映射和路径检查逻辑
        start_node = sum(int(c) for c in target_str) % 2
        end_node = 1 - start_node
        paths = list(nx.shortest_simple_paths(G, start_node, end_node))
        
        if len(paths) != 1 or not path_matches_string(paths[0], target_str):
            continue
        
        local_added.add(target_str)
        save_path(paths[0], target_str)

# 复用前面的path_matches_string和save_path函数

def main():
    G = nx.Graph()
    G.add_edges_from([(0,2), (2,1), (0,3), (3,1)])
    
    all_strings = [format(i, '03b') for i in range(8)]
    
    # 预划分镜像任务组，去重
    seen = set()
    task_groups = []
    current_group = []
    group_size = 2  # 每组任务数，可根据CPU核心数调整
    
    for s in all_strings:
        mirror = s[::-1]
        if s not in seen and mirror not in seen:
            current_group.append(s)
            seen.add(s)
            seen.add(mirror)
            
            if len(current_group) == group_size:
                task_groups.append(current_group)
                current_group = []
    if current_group:
        task_groups.append(current_group)
    
    # 分配任务给进程池
    with mp.Pool(processes=len(task_groups)) as pool:
        pool.starmap(worker, [(group, G) for group in task_groups])

if __name__ == "__main__":
    main()

适用场景 #

适合大规模二进制串场景（比如n位串，n很大），完全没有进程间通信和锁竞争，效率最高。预处理镜像对时可以用frozenset({s, s[::-1]})作为键来分组，确保高效去重。

方案三：主进程负责状态同步（适合路径生成成本高的场景） #

如果生成路径的计算成本很高（比如图结构复杂、最短路径计算耗时久），可以让子进程专注于生成路径，把结果提交给主进程，由主进程统一维护added集合并判断是否符合约束。

实现思路 #

1. 主进程创建任务队列和结果队列，把所有二进制串放到任务队列。
2. 子进程从任务队列取任务，生成路径后把结果发送到结果队列。
3. 主进程从结果队列取出结果，统一检查是否违反约束（是否已存在镜像串、路径是否唯一匹配），然后更新added集合。

代码示例 #

import multiprocessing as mp
import networkx as nx
from queue import Empty

def worker(task_queue, result_queue, G):
    while True:
        try:
            target_str = task_queue.get(timeout=1)
        except Empty:
            break
        
        # 生成路径（替换成你的实际逻辑）
        start_node = sum(int(c) for c in target_str) % 2
        end_node = 1 - start_node
        paths = list(nx.shortest_simple_paths(G, start_node, end_node))
        
        if len(paths) == 1 and path_matches_string(paths[0], target_str):
            result_queue.put((target_str, paths[0]))

# 复用前面的path_matches_string和save_path函数

def main():
    G = nx.Graph()
    G.add_edges_from([(0,2), (2,1), (0,3), (3,1)])
    
    all_strings = [format(i, '03b') for i in range(8)]
    
    # 初始化队列
    task_queue = mp.Queue()
    result_queue = mp.Queue()
    
    for s in all_strings:
        task_queue.put(s)
    
    # 启动子进程
    processes = []
    for _ in range(mp.cpu_count()):
        p = mp.Process(target=worker, args=(task_queue, result_queue, G))
        p.start()
        processes.append(p)
    
    # 主进程处理结果，维护状态
    added = set()
    while True:
        try:
            target_str, path = result_queue.get(timeout=2)
        except Empty:
            # 所有子进程都结束则退出循环
            if all(not p.is_alive() for p in processes):
                break
            continue
        
        mirror_str = target_str[::-1]
        if target_str in added or mirror_str in added:
            continue
        
        added.add(target_str)
        save_path(path, target_str)
    
    # 等待所有子进程结束
    for p in processes:
        p.join()

if __name__ == "__main__":
    main()

适用场景 #

适合路径生成计算密集的场景，子进程专注于计算，主进程只做轻量的约束检查，最大化利用CPU资源。

通用注意事项 #

1. NetworkX图的传递：如果图是只读的，可以直接传给子进程（Python会序列化传递）；如果需要修改图，则需要用multiprocessing.Manager创建共享图对象。
2. 进程数设置：通常设置为CPU核心数（mp.cpu_count()），避免过多进程导致上下文切换开销。
3. 大规模串优化：对于n位二进制串，预划分镜像对时可以用字典按"串和镜像的最小字典序"作为键来分组，提升预处理效率。

内容的提问来源于stack exchange，提问作者Tomasz Przemski