如何在运行脚本时将更多项目添加到多处理队列中 [英] how to add more items to a multiprocessing queue while script in motion

问题描述

我正在尝试通过队列学习多处理.

I am trying to learn multiprocessing with queue.

我想做的是弄清楚脚本运行时何时/如何将更多项目添加到队列中".

What I want to do is figure out when/how to "add more items to the queue" when the script is in motion.

以下脚本是我正在使用的基准:

The below script is the baseline I am working from:

import multiprocessing


class MyFancyClass:

    def __init__(self, name):
        self.name = name

    def do_something(self):
        proc_name = multiprocessing.current_process().name
        print('Doing something fancy in {} for {}!'.format(
            proc_name, self.name))


def worker(q):
    obj = q.get()
    obj.do_something()


if __name__ == '__main__':
    queue = multiprocessing.Queue()

    p = multiprocessing.Process(target=worker, args=(queue,))
    p.start()

    queue.put(MyFancyClass('Fancy Dan'))
    queue.put(MyFancyClass('Frankie'))
    print(queue.qsize())

    # Wait for the worker to finish
    queue.close()
    queue.join_thread()
    p.join()

在第26行的

上，Fancy Dan注入有效，但Frankie无效.我能够确认Frankie确实进入了队列.我需要一个可以检查更多项目"并将其根据需要插入队列的位置.如果不存在更多项目，请在清除现有项目后关闭队列.

on line 26, the Fancy Dan inject works, but the Frankie piece doesn't. I am able to confirm that Frankie does make it into the queue. I need a spot where I can "Check for more items" and insert them into the queue as needed. If no more items exist, then close the queue when the existing items are clear.

我该怎么做?

谢谢！

推荐答案

让我们弄清楚:

• 目标函数worker(q)在上述方案中仅被调用一次.在第一次调用时，该函数将暂停等待阻塞操作q.get()的结果.它从queue获取实例MyFancyClass('Fancy Dan')，调用其do_something方法并完成操作.
• MyFancyClass('Frankie')将被放入队列中，但不会进入Process，因为该进程的目标功能已完成.
• 其中一种方法是从队列中读取并等待一个信号/标记的项目，该信号/信号表明已停止使用队列.假设None值.

• the target function worker(q) will be called just once in the above scheme. At that first call the function will suspend waiting the result from blocking operation q.get(). It gets the instance MyFancyClass('Fancy Dan') from the queue, invokes its do_something method and get finished.
• MyFancyClass('Frankie') will be put into the queue but won't go to the Process cause the process' target function is done.
• one of the ways is to read from the queue and wait for a signal/marked item which signals that queue usage is stopped. Let's say None value.

import multiprocessing


class MyFancyClass:

    def __init__(self, name):
        self.name = name

    def do_something(self):
        proc_name = multiprocessing.current_process().name
        print('Doing something fancy in {} for {}!'.format(proc_name, self.name))


def worker(q):
    while True:
        obj = q.get()
        if obj is None:
            break
        obj.do_something()


if __name__ == '__main__':
    queue = multiprocessing.Queue()

    p = multiprocessing.Process(target=worker, args=(queue,))
    p.start()

    queue.put(MyFancyClass('Fancy Dan'))
    queue.put(MyFancyClass('Frankie'))
    # print(queue.qsize())
    queue.put(None)

    # Wait for the worker to finish
    queue.close()
    queue.join_thread()
    p.join()

输出:

Doing something fancy in Process-1 for Fancy Dan!
Doing something fancy in Process-1 for Frankie!

这篇关于如何在运行脚本时将更多项目添加到多处理队列中的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！