Python使用multiprocessing生成排列时如何去除重复结果

• 当前并行逻辑未做任务分片，启动的每个工作进程都会独立执行完整的permutations全量遍历流程，所有进程的任务范围完全重叠，最终生成的排列会被每个进程重复写入日志，重复次数恰好为cores-1。
• 多进程场景下多个进程同时以追加模式写同一个Logs.txt，还会附带出现写入竞争、日志行错乱的问题。

核心修改点 #

1. 拆分任务分片：利用itertools.islice对排列生成器做切片切分，为每个进程分配专属、无重叠的处理范围，从根源上杜绝任务重叠。
2. 隔离文件写入：通过进程安全的队列把待写入内容传递给单独的写工作进程统一落盘，避免多进程同时操作同一文件带来的竞争问题。
3. 优化内存占用：避免将全量数据库集合作为参数传给每个子进程，减少多进程启动时的冗余内存拷贝。

import sys
import os
import pickle
import multiprocessing
from itertools import permutations, islice

DATABASE = r'C:/Python38/dataSource/'
# 待生成排列的元素列表
items = ['apple', 'ball', 'cat', 'dog', 'elephant', 'fish', 'goat', 'horse', 'jaguar', 'kitten', 'leopard', 'monkey', 'nest']
cores = 3
# 提前计算总排列数：13个元素取3的全排列共13*12*11=1716条
TOTAL_PERM_COUNT = len(items) * (len(items)-1) * (len(items)-2)

# 独立写文件的工作进程：统一处理所有进程的日志写入
def file_writer(write_queue):
    with open("Logs.txt", "a", encoding="utf-8") as f:
        for line in iter(write_queue.get, None):
            f.write(f"{line}\n")

# 子进程执行的排列生成、比对任务
def perm_worker(task_start, task_end, database, write_queue):
    process_count = 0
    # 切出当前进程负责的排列片段，不遍历全量序列
    perm_segment = islice(permutations(items, 3), task_start, task_end)
    for word_tuple in perm_segment:
        word_str = ' '.join(word_tuple)
        # 在此处补充原计划与pickle数据库的比对逻辑，判断是否符合匹配规则
        # 示例：将不在数据库集合中的排列写入日志
        if word_str not in set().union(*database):
            write_queue.put(word_str)
        process_count += 1
        sys.stdout.write(f'当前进程处理进度: {process_count}/{task_end - task_start}\r')

if __name__ == '__main__':
    # 加载pickle数据库的逻辑保持不变
    database = [set() for _ in range(4)]
    file_total = len(os.listdir(DATABASE))
    half = file_total // 2
    quarter = half // 2
    for idx, file_name in enumerate(os.listdir(DATABASE)):
        print(f'\rReading database: {idx + 1}/{file_total}', end=' ')
        with open(DATABASE + file_name, 'rb') as f:
            data = pickle.load(f)
            if idx < half:
                if idx < quarter:
                    database[0] = database[0] | data
                else:
                    database[1] = database[1] | data
            else:
                if idx < half + quarter:
                    database[2] = database[2] | data
                else:
                    database[3] = database[3] | data
    print(f'\nDATABASE PATH => {DATABASE}')

    # 启动独立写文件进程
    write_queue = multiprocessing.Queue()
    writer_proc = multiprocessing.Process(target=file_writer, args=(write_queue,))
    writer_proc.start()

    # 计算每个进程的任务边界，确保所有排列无重叠、无遗漏
    jobs = []
    step = TOTAL_PERM_COUNT // cores
    for proc_idx in range(cores):
        start_pos = proc_idx * step
        # 最后一个进程处理到序列末尾，吸收分片时的余数，避免遗漏排列
        end_pos = TOTAL_PERM_COUNT if proc_idx == cores -1 else (proc_idx + 1) * step
        worker_proc = multiprocessing.Process(
            target=perm_worker,
            args=(start_pos, end_pos, database, write_queue)
        )
        jobs.append(worker_proc)
        worker_proc.start()

    # 等待所有计算进程执行完毕
    for p in jobs:
        p.join()
    # 发送结束信号，退出写文件进程
    write_queue.put(None)
    writer_proc.join()

• 无论设置多少个核心，每个排列只会被对应分片的进程生成、处理1次，完全消除重复写入的问题
• 独立的写进程统一操作日志文件，不会出现多进程写文件导致的内容错乱、覆盖问题

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