文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Queue队列中join()与task_done()的关系及说明

2023-02-25 11:20

关注

join()与task_done()的关系

在网上大多关于join()与task_done()的结束原话是这样的:

但是可能很多人还是不太理解,这里以我自己的理解来阐述这两者的关联。

理解

如果线程里每从队列里取一次,但没有执行task_done(),则join无法判断队列到底有没有结束,在最后执行个join()是等不到结果的,会一直挂起。

可以理解为,每task_done一次 就从队列里删掉一个元素,这样在最后join的时候根据队列长度是否为零来判断队列是否结束,从而执行主线程。

下面看个自己写的例子:

下面这个例子,会在join()的地方无限挂起,因为join在等队列清空,但是由于没有task_done,它认为队列还没有清空,还在一直等。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
'''threading test'''
import threading
import queue
from time import sleep
#之所以为什么要用线程,因为线程可以start后继续执行后面的主线程,可以put数据,如果不是线程直接在get阻塞。
class Mythread(threading.Thread):
 def __init__(self,que):
 threading.Thread.__init__(self)
 self.queue = que
 def run(self):
 while True:
 sleep(1)
 if self.queue.empty(): #判断放到get前面,这样可以,否则队列最后一个取完后就空了,直接break,走不到print
 break
 item = self.queue.get()
 print(item,'!')
 #self.queue.task_done()
 return
que = queue.Queue()
tasks = [Mythread(que) for x in range(1)]
for x in range(10):
 
 que.put(x) #快速生产
for x in tasks:
 t = Mythread(que) #把同一个队列传入2个线程
 t.start()
 
que.join()
 
print('---success---')

如果把self.queue.task_done()  注释去掉,就会顺利执行完主程序。

这就是“ Queue.task_done()函数向任务已经完成的队列发送一个信号”这句话的意义,能够让join()函数能判断出队列还剩多少,是否清空了。

而事实上我们看下queue的源码可以看出确实是执行一次未完成队列减一:

 def task_done(self):
 '''Indicate that a formerly enqueued task is complete.
 Used by Queue consumer threads. For each get() used to fetch a task,
 a subsequent call to task_done() tells the queue that the processing
 on the task is complete.
 If a join() is currently blocking, it will resume when all items
 have been processed (meaning that a task_done() call was received
 for every item that had been put() into the queue).
 Raises a ValueError if called more times than there were items
 placed in the queue.
 '''
 with self.all_tasks_done:
 unfinished = self.unfinished_tasks - 1
 if unfinished <= 0:
 if unfinished < 0:
 raise ValueError('task_done() called too many times')
 self.all_tasks_done.notify_all()
 self.unfinished_tasks = unfinished
 

快速生产-快速消费

上面的演示代码是快速生产-慢速消费的场景,我们可以直接用task_done()与join()配合,来让empty()判断出队列是否已经结束。

当然,queue我们可以正确判断是否已经清空,但是线程里的get队列是不知道,如果没有东西告诉它,队列空了,因此get还会继续阻塞,那么我们就需要在get程序中加一个判断,如果empty()成立,break退出循环,否则get()还是会一直阻塞。

慢速生产-快速消费

但是如果生产者速度与消费者速度相当,或者生产速度小于消费速度,则靠task_done()来实现队列减一则不靠谱,队列会时常处于供不应求的状态,常为empty,所以用empty来判断则不靠谱。

那么这种情况会导致 join可以判断出队列结束了,但是线程里不能依靠empty()来判断线程是否可以结束。

我们可以在消费队列的每个线程最后塞入一个特定的“标记”,在消费的时候判断,如果get到了这么一个“标记”,则可以判定队列结束了,因为生产队列都结束了,也不会再新增了。

代码如下:

#!/usr/bin/env python
# -*- coding:utf-8 -*-
'''threading test'''
import threading
import queue
from time import sleep
#之所以为什么要用线程,因为线程可以start后继续执行后面的主线程,可以put数据,如果不是线程直接在get阻塞。
class Mythread(threading.Thread):
 def __init__(self,que):
 threading.Thread.__init__(self)
 self.queue = que
 def run(self):
 while True:
 item = self.queue.get()
 self.queue.task_done() #这里要放到判断前,否则取最后最后一个的时候已经为空,直接break,task_done执行不了,join()判断队列一直没结束
 if item == None:
 break
 print(item,'!')
return
que = queue.Queue()
tasks = [Mythread(que) for x in range(1)]
 #快速生产
for x in tasks:
 t = Mythread(que) #把同一个队列传入2个线程
 t.start()
for x in range(10):
 sleep(1)
 que.put(x)
for x in tasks:
 que.put(None)
que.join()
print('---success---')

注意点

put队列完成的时候千万不能用task_done(),否则会报错:

task_done() called too many times

因为该方法仅仅表示get成功后,执行的一个标记。

总结

以上为个人经验,希望能给大家一个参考,也希望大家多多支持编程网。

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯