清空队列时出现问题

Question

问题:

我是多道处理的新手，无论我尝试了什么，我都一事无成。每次我想我发现了什么，我就遇到了新的障碍。我的目标是使用多个进程加载队列，然后使用多个进程从队列中提取数据并处理数据。我已经尝试恢复到基本的队列处理，但一旦实现了多个进程，就无法从队列中获得任何东西。我错过了什么

代码

rom multiprocessing import Process, Lock
from queue import Queue
import os

q = Queue(5)


def get_from_q():
    print('trying to get')
    print(q.get())


if __name__ == '__main__':

    # put items at the end of the queue
    for x in range(6):
        print('adding ', x)
        q.put(x)

    PROCESSOR_COUNT = os.cpu_count()
    processes = []
    for p in range(PROCESSOR_COUNT):
        print('spawning process')
        p = Process(target=get_from_q)
        processes.append(p)

    for p in processes:
        print('starting')
        p.start()

    for p in processes:
        print('joining')
        p.join()

结果：

    adding 0
    adding 1
    adding 2
    adding 3
    adding 4
    adding 5

预期结果

    adding 0
    adding 1
    adding 2
    adding 3
    adding 4
    adding 5
    spawning process
    spawning process
    spawning process
    spawning processv
    starting
    starting
    starting
    starting
    trying to get 
    0
    trying to get 
    1 
    trying to get 
    2 
    trying to get 
    3 
    trying to get 
    4
    trying to get 
    5
    joining
    joining
    joining
    joining

Answer 1

如果您在使用spawn创建新进程的平台下运行，那么在创建新进程时，将不会继承主进程的地址空间，而是通过重新执行程序顶部的所有代码来初始化新的地址空间。这意味着您在全局范围内定义的任何内容都将重新执行，例如在您的代码中：

q = Queue(5)

这意味着此代码由您创建的每个进程执行，这意味着每个进程都有自己的副本q。这是行不通的。您需要创建一次q，并将其作为参数传递。我还向print函数添加了flush=True，以减少不同进程的输出被交错的可能性

from multiprocessing import Process, Lock, Queue
import os


def get_from_q(q):
    print('trying to get', q.get(), flush=True)


if __name__ == '__main__':
    PROCESSOR_COUNT = os.cpu_count()

    q = Queue(PROCESSOR_COUNT) # or put no size limitation on this

    # put items at the end of the queue
    for x in range(PROCESSOR_COUNT):
        print('adding ', x)
        q.put(x)

    processes = []
    for p in range(PROCESSOR_COUNT):
        print('spawning process')
        p = Process(target=get_from_q, args=(q,))
        processes.append(p)

    for p in processes:
        print('starting', flush=True)
        p.start()

    for p in processes:
        print('joining', flush=True)
        p.join()

印刷品：

adding  0
adding  1
adding  2
adding  3
adding  4
adding  5
adding  6
adding  7
spawning process
spawning process
spawning process
spawning process
spawning process
spawning process
spawning process
spawning process
starting
starting
starting
starting
starting
starting
starting
starting
joining
trying to get 0
trying to get 1
trying to get 2
trying to get 3
trying to get 4
trying to get 5
trying to get 6
joining
joining
joining
trying to get 7
joining
joining
joining
joining

使用进程池

此处，队列由池实现隐藏：

from multiprocessing import Pool, cpu_count


def worker(x):
    print('x =', x, flush=True)
    return x ** 2


if __name__ == '__main__':
    PROCESSOR_COUNT = cpu_count()

    pool = Pool(PROCESSOR_COUNT) #
    print(pool.map(worker, range(PROCESSOR_COUNT)))

印刷品：

x = 0
x = 1
x = 2
x = 3
x = 4
x = 5
x = 6
x = 7
[0, 1, 4, 9, 16, 25, 36, 49]