在worker方法中动态添加对象到队列答案

【问题标题】：Dynamically add object to queue inside the worker method在worker方法中动态添加对象到队列
【发布时间】：2021-09-22 17:23:21
【问题描述】：

我正在使用 python 多处理池和队列来并行运行任务。但我必须在队列中动态插入其他作业并等待它们完成（他们也可以在队列中插入其他作业）。

def add_another(q,blocked_name):
    name = q.get()
    if 'a' in name and name not in blocked_name:
        print('The name contains an a')
        # Here I want to add another name in the queue
        # Like q.put('Another') even if 'Another' will create a loop

if __name__ == '__main__':

    m = Manager()
    q = m.Queue()
    p = Pool(3)

    worker = []

    req = ['alice','amanda','bob','mallery']

    for d in req:
        q.put(d)

    blocked_name = ['mallery','steve']

    for i in range(len(req)):
        worker.append(p.apply_async(add_another, (q,blocked_name,)))

    # Here I want to wait ALL the worker, even the one added inside the add_another method 
    [r.get() for r in worker]

我该怎么做？

感谢您的帮助

【问题讨论】：

您的目标是从一个初始子进程中添加一个额外的工作进程吗？

标签： python multithreading queue threadpool

【解决方案1】：

您似乎将 Queue 和工作进程的内容混为一谈——因为简单地为前者添加一个名称并不一定会在池工作进程中应用该函数。

做你想做的事情的一种方法是利用可选的callback 函数apply_async() 支持，并让它在更新时从队列中创建另一个工作进程。这可确保为添加到队列中的每个作业创建一个。

这就是我的建议。请注意，我更改了一些变量的名称以使代码更具可读性。

from multiprocessing import Manager, Pool


def add_another(queue, blocked_names):
    name = queue.get()
    print(f'processing {name}')
    if 'a' in name and name not in blocked_names:
        print('The name contains an "a"')
        queue.put('Another')  # Add another to queue.
        return True  # Indicate another was added.
    return False  # Indicate another one NOT added.

def check_result(another_added):
    ''' Start processing of another task if one was added to the queue. '''
    if another_added:
        results.append(pool.apply_async(add_another, (queue, blocked_names),
                                        callback=check_result))

if __name__ == '__main__':

    mgr = Manager()
    queue = mgr.Queue()
    pool = Pool(3)

    results = []
    reqs = ['alice', 'amanda', 'bob', 'mallery']
    blocked_names = ['mallery', 'steve']

    # Initialize task queue.
    for req in reqs:
        queue.put(req)

    for _ in range(len(reqs)):
        results.append(pool.apply_async(add_another, (queue, blocked_names),
                                        callback=check_result))

    res = [result.get() for result in results]
#    print(f'{res=}')
    print('-Fini-')

【讨论】：