如何将项目放回队列.Queue

2023年1月29日 164次阅读

如何将项目返回队列.问题？如果任务失败,这在线程或多处理中很有用,这样任务就不会丢失.

docs for queue.Queue.get()说该函数可以“从队列中删除并返回一个项目”,但我相信这里使用“return”这个词指的是将该项返回给调用线程的函数,而不是将它放回到项目队列中.下面的示例代码只是在主线程的第二个queue.Queue.get()调用上无限地阻塞,而不是在线程中调用print()调用.

import time
import threading
import queue


def threaded_func():
    thread_task = myqueue.get()
    print('thread_task: ' + thread_task)

myqueue = queue.Queue()
myqueue.put('some kind of task')
main_task = myqueue.get()
print('main_task: ' + main_task)

t = threading.Thread(target=threaded_func)
t.daemon = True
t.start()

time.sleep(5)
myqueue.get()   # This blocks indefinitely

我不得不相信有一种简单的方法可以让任务恢复,那又是什么呢？调用task_done()然后使用任务put()将它放回到队列中的两个操作不是原子的,因此可能导致丢失的项目.

一个可能但又笨重的解决方案是尝试再次执行任务,但是你必须添加一些额外的行来处理这种复杂性,我甚至不确定所有失败的任务是否必然会在那里恢复办法.

最佳答案 并非所有失败的任务都可以恢复你不应该重试它们,除非有理由认为它们会在以后通过.例如,如果您的工作项是URL并且连接失败计数,则可以实现某种最大重试次数.

你最大的问题是你还没有实现可行的工人模型.您需要2个队列才能与工作人员进行双向对话.一个用于发布工作项,一个用于接收状态.一旦你有了,接收器总是可以决定将该消息塞回到工作队列中.这是一个懒惰的工作人员的例子,它只是传递了它所说的.

import threading
import queue

def worker(in_q, out_q):
    while True:
        try:
            task, data = in_q.get()
            print('worker', task, data)
            if task == "done":
                return
            elif task == "pass this":
                out_q.put(("pass", data))
            else:
                out_q.put(("fail", data))
        except Exception as e:
            print('worker exception', e)
            out_q.put("exception", data)

in_que = queue.Queue()
out_que = queue.Queue()

work_thread = threading.Thread(target=worker, args=(in_que, out_que))
work_thread.start()

# lets make every other task a fail
in_que.put(('pass this', 0))
in_que.put(('fail this', 1))
in_que.put(('pass this', 2))
in_que.put(('fail this', 3))
in_que.put(('pass this', 4))
in_que.put(('fail this', 5))

pending_tasks = 6

while pending_tasks:
    status, data = out_que.get()
    if status == "pass":
        pending_tasks -= 1
    else:
        # make failing tast pass
        in_que.put(('pass this', data))

in_que.put(("done", None))
work_thread.join()
print('done')
点赞