多进程质数查找脚本进程交替异常终止问题排查求助

你的问题核心是任务划分导致的负载严重不均衡，直接造成了半数进程快速退出的现象，我们一步步拆解分析：

为什么会出现交替终止？ #

你给8个进程分配的初始值是1到8，步长为8，这意味着：

• 偶数初始的进程（2、4、6、8）：只处理偶数序列，而除了2之外，所有偶数都是合数。这些进程每次循环只需要检查i%2 == 0就能判定为合数，然后快速递增i，直到超过maxnum后立刻终止——这就是为什么半数进程会快速结束。
• 奇数初始的进程（1、3、5、7）：需要处理所有奇数序列，其中包含大量质数，必须遍历到√i才能确定是否为质数，耗时极长，所以会持续运行。

加上进程是按顺序启动的（1→2→3→4…），偶数进程启动后秒退，奇数进程持续运行，就呈现出了“启动、终止、启动、终止”的交替模式。

修复方案：让所有进程负载均衡 #

方案1：聚焦奇数任务（最直接高效） #

既然偶数除了2都是合数，我们可以先单独处理2，然后让所有进程只处理奇数，这样每个进程的工作量大致相同，不会出现闲忙不均的情况。

修改后的代码：

import time
import multiprocessing

def prime(start_i, step, maxnum):
    primes = []
    i = start_i
    while i <= maxnum:
        is_prime = True
        # 奇数无需检查偶数因数，从3开始步长2，减少一半循环次数
        for j in range(3, int(i**0.5) + 1, 2):
            if i % j == 0:
                is_prime = False
                break
        if is_prime:
            primes.append(i)
        i += step
    return primes

def main():
    start = time.monotonic()
    maxnum = 10000000
    processes = 8
    all_primes = []
    
    # 先处理唯一的偶质数2
    if maxnum >= 2:
        all_primes.append(2)
    
    # 用进程池管理任务，每个进程处理不同起始的奇数序列
    with multiprocessing.Pool(processes) as pool:
        results = []
        # 起始值为1、3、5...15，步长为16（2*processes），覆盖所有奇数
        for i in range(processes):
            start_i = 1 + 2*i
            results.append(pool.apply_async(prime, args=(start_i, 2*processes, maxnum)))
        
        # 收集所有进程的结果
        for res in results:
            all_primes.extend(res.get())
    
    all_primes.sort()
    print(f"time taken: {(time.monotonic() - start)/60:.2f} mins")
    print(f"Total primes found: {len(all_primes)}")

if __name__ == "__main__":
    main()

方案2：动态任务队列（更灵活） #

如果需要应对更复杂的任务场景，或者希望动态分配工作负载，可以用队列来分发待检查的数字，每个进程从队列取数处理，直到队列为空。

示例代码：

import time
import multiprocessing
from queue import Empty

def prime_worker(task_queue, result_queue):
    while True:
        try:
            i = task_queue.get(timeout=1)
        except Empty:
            break
        if i <= 1:
            continue
        is_prime = True
        for j in range(2, int(i**0.5)+1):
            if i % j == 0:
                is_prime = False
                break
        if is_prime:
            result_queue.put(i)

def main():
    start = time.monotonic()
    maxnum = 10000000
    processes = 8
    
    # 创建任务队列和结果队列
    task_queue = multiprocessing.Queue()
    result_queue = multiprocessing.Queue()
    
    # 只放入需要检查的数：2 + 所有奇数
    task_queue.put(2)
    for i in range(3, maxnum+1, 2):
        task_queue.put(i)
    
    # 启动进程
    plist = []
    for _ in range(processes):
        p = multiprocessing.Process(target=prime_worker, args=(task_queue, result_queue))
        p.start()
        plist.append(p)
    
    # 等待所有进程完成
    for p in plist:
        p.join()
    
    # 收集结果
    all_primes = []
    while not result_queue.empty():
        all_primes.append(result_queue.get())
    all_primes.sort()
    
    print(f"time taken: {(time.monotonic() - start)/60:.2f} mins")
    print(f"Total primes found: {len(all_primes)}")

if __name__ == "__main__":
    main()

额外优化提示 #

• 避免使用manager.list()：进程间共享列表的通信开销极大，改用进程返回结果或结果队列，效率会提升很多。
• 质数判断优化：对于奇数，只检查奇数因数（从3开始步长2），能减少一半的循环次数。
• 用multiprocessing.Pool：比手动管理进程更简洁，自带进程复用和任务调度功能。

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