如何用Python多进程计算质数？树莓派3a+多核优化需求

你的单进程质数计算代码只能利用一个CPU核心，而树莓派3A+是四核，完全可以通过多进程来大幅提升计算速度。不推荐用_thread模块——Python的全局解释器锁（GIL）会限制线程在CPU密集型任务中的并行能力，真正能利用多核的是multiprocessing库。

实现思路 #

1. 主进程负责读取初始质数列表、管理任务分发、更新共享质数集合、定期保存数据
2. 多个工作进程异步处理质数判断任务，通过队列传递待检查数字和结果
3. 用共享列表加锁的方式，保证多进程安全访问已发现的质数

修改后的代码 #

print("Importing Libraries...")
import pandas as pd
import csv
import math
from multiprocessing import Process, Manager, Queue, Lock
import time

def is_prime(num, primes_snapshot):
    """判断指定数字是否为质数，使用质数列表快照避免锁竞争"""
    if num < 2:
        return False
    num_sqrt = math.floor(math.sqrt(num))
    for p in primes_snapshot:
        if p > num_sqrt:
            break
        if num % p == 0:
            return False
    return True

def worker(task_queue, result_queue, primes_list, lock):
    """工作进程：从任务队列取数，判断质数后返回结果"""
    while True:
        num = task_queue.get()
        if num is None:  # 收到终止信号，退出进程
            break
        # 获取当前质数列表的快照，减少锁持有时间
        with lock:
            primes_snapshot = primes_list.copy()
        if is_prime(num, primes_snapshot):
            result_queue.put(num)

def main():
    # 读取初始质数文件
    print("Reading file...")
    # 替换为树莓派本地的文件路径，确保有读写权限
    file = open("/home/pi/primes.csv", "r")
    data = list(csv.reader(file, delimiter=","))
    file.close()
    print("Done.")

    # 转换为整数质数列表
    print("Converting to integer list...")
    prime_str = [row[1] for row in data]
    prime_str.pop(0)  # 移除表头行的"0"
    primes = list(map(int, prime_str))
    print("Done.")

    # 初始化计数器和起始数字
    n = len(primes) - 1
    N_START = n
    current_num = primes[-1] + 1
    print(f"Starting from number {current_num}, current prime count: {n}")

    # 创建多进程共享资源
    manager = Manager()
    primes_list = manager.list(primes)  # 多进程共享的质数列表
    lock = Lock()  # 保护质数列表修改的锁
    task_queue = Queue(maxsize=100)  # 待检查数字的任务队列
    result_queue = Queue()  # 新发现质数的结果队列

    # 启动工作进程（树莓派3A+是四核，建议启动3个进程避免占满系统资源）
    num_workers = 3
    workers = []
    for _ in range(num_workers):
        p = Process(target=worker, args=(task_queue, result_queue, primes_list, lock))
        p.start()
        workers.append(p)

    try:
        while True:
            # 向任务队列添加待检查的数字
            task_queue.put(current_num)
            current_num += 1

            # 处理结果队列中的新质数
            while not result_queue.empty():
                new_prime = result_queue.get()
                n += 1
                print(f"{new_prime} is the {n}th prime number")
                # 安全更新共享质数列表
                with lock:
                    primes_list.append(new_prime)
                # 每75000个质数保存一次文件
                if n % 75000 == 0:
                    print("Saving primes to file...")
                    primes_df = pd.DataFrame(list(primes_list))
                    primes_df.to_csv("/home/pi/primes.csv", index_label="index")
                    print("Save done.")

            # 短暂休眠，避免主进程过度占用CPU
            time.sleep(0.001)

    except KeyboardInterrupt:
        print("\nReceived interrupt, stopping workers...")
        # 向每个工作进程发送终止信号
        for _ in range(num_workers):
            task_queue.put(None)
        # 等待所有工作进程结束
        for p in workers:
            p.join()

    finally:
        # 程序退出前最终保存所有质数
        print("Final save...")
        primes_df = pd.DataFrame(list(primes_list))
        primes_df.to_csv("/home/pi/primes.csv", index_label="index")
        print(f"\nCalculated and backed {n} prime numbers ({n - N_START} new prime numbers)")

if __name__ == "__main__":
    main()

关键细节说明 #

• 共享资源安全：用Manager.list创建跨进程共享的质数列表，配合Lock确保多个进程不会同时修改列表，避免数据混乱
• 任务队列：主进程持续生成待检查数字，工作进程异步取任务，实现负载均衡
• 质数判断优化：工作进程每次判断时获取质数列表的快照，减少锁的持有时间，提升并发效率
• 文件保存：统一由主进程负责写入CSV文件，避免多进程同时写文件导致的数据损坏
• 进程终止：捕获Ctrl+C中断时，向工作进程发送None作为终止信号，确保进程正常退出

树莓派适配注意事项 #

• 树莓派3A+是四核CPU，建议启动3个工作进程（留一个核心给系统后台任务）
• 把代码中的文件路径替换为树莓派本地路径，比如/home/pi/primes.csv，并确保当前用户有该路径的读写权限

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