关于python 多进程异步打印相同值的问题

Question

我使用该程序，创建了3个进程的线程池，但是打印结果出现了，队列容量重复的问题：

这个该怎么解决
代码如下:

from multiprocessing import Pool, Manager, cpu_count
import time


num_list = []

for i in range(10):
    num_list.append(i)

start_time = time.time()
def crawler(q, index):
    Process_id = "Process"+ str(index)
    while not q.empty():
        time.sleep(1)
        num = q.get(timeout=1)
        print(Process_id, num, q.qsize())


if __name__ == '__main__':
    manager = Manager()
    num_queue = manager.Queue(150000)

    # 填充队列
    for num in num_list:
        num_queue.put(num)

    lock = manager.Lock()
    # 创建线程池为10个
    pool = Pool(processes=3)
    # 创建的任务数量
    for i in range(3):
        pool.apply_async(func=crawler, args=(num_queue, i))

    print("Started procesess")
    pool.close()
    pool.join()

    print("Main Process end")
    end_time = time.time()
    print("时间为"+str(end_time-start_time))

Answer 1

以下回答参考 皆我百晓生、券券喵儿 等免费微信小程序相关内容作答，并由本人整理回复。

这个问题是由于多进程之间的通信问题导致的。在Python中，我们不能直接共享内存或文件来实现多进程间的通信。为了防止出现类似的问题，你需要确保你的数据结构和算法设计能够支持多进程并行处理。

以下是一个可能的解决方案：

from multiprocessing import Pool, Manager, cpu_count
import time

# 创建一个全局变量来存储所有的数字
all_numbers = [i for i in range(10)]

# 创建一个全局队列来存放所有数字
global_num_queue = Manager().Queue()

# 将所有数字放入队列
for number in all_numbers:
    global_num_queue.put(number)

# 创建一个Pool对象，并指定要使用的CPU核心数
pool = Pool(cpu_count())

# 定义一个函数，它会从队列中取出一个元素并打印出来
def worker():
    while True:
        try:
            number = global_num_queue.get(block=True, timeout=1)
            print(f"Process {number}", end=" ")

        except Exception as e:
            print(e)
            break

# 使用apply_async方法启动3个子任务
for _ in range(3):
    pool.apply_async(worker)

print("\nStarted processes\n")

# 等待所有子任务完成
pool.close()
pool.join()

print("\nMain process end\n")
end_time = time.time()
print(f"time taken: {end_time - start_time}")

在这个解决方案中，我们创建了一个全局变量all_numbers来存储所有的数字，并将它们放入了一个全局队列global_num_queue中。然后，我们创建了一个Pool对象，并指定了要使用的CPU核心数。接着，我们定义了一个函数worker，这个函数会在主进程中阻塞地等待队列中的元素被获取到。最后，我们通过apply_async方法启动了3个子任务。

这样，当子任务开始执行时，它们可以同时访问队列中的元素，而不会引起其他子任务的阻塞。