多线程中的 Queue队列中join()与task_done() ,

Posted 2020-11-19 dahuag

Queue.task_done() 在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号

Queue.join() 实际上意味着等到队列为空，再执行别的操作

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

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

下面看个自己写的例子：

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

class BdSpider(threading.Thread):
    """搜狗移动相关词获取爬虫"""
    def __init__(self, waiting):
        super(BdSpider, self).__init__()
        self.waiting = waiting
        self.headers = {
                ‘User-Agent‘: "Mozilla/5.0 (Linux; android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (Khtml, like Gecko) Chrome/66.0.3359.139 Mobile Safari/537.36"
            }

    def run(self):
        flag = True
        while flag:
            ipone = self.waiting.get()
            error = self.get_html(ipone)
            if error == -1:
                flag = False
            # self.waiting.task_done()

    def get_html(self, ipone,num_retries=3):
        global allipus
        global error_counter
        error_counter = 0
        try:
            ip = {"http":"http://"+str(ipone) , "https":"http://"+str(ipone)}
            requests.get(‘https://m.sogou.com/‘, proxies=ip , timeout = 3 , headers = self.headers)
        except:
            error_counter += 1
            if num_retries > 0:
                return self.get_html(ipone , num_retries - 1)
        else:
            if str(ipone) not in str(allipus):
                print("保留ip：" + str(ipone))
                allipus.append(ipone)
            error_counter = 0
        if error_counter > 50:
            return -1

#更新获取ip
def ipAction(ipsss):
    ipall = []
    if ipsss:
        for iss in ipsss:
            if iss:
                ipall.append(iss)
    url = "http://api.ip.data5u.com/api/get.shtml?order=194469bc039df0eb7580e09541aa0624&num=20&carrier=0&protocol=2&an1=1&an2=2&an3=3&sp1=1&sp2=2&sp3=3&sort=1&system=1&distinct=0&rettype=1&seprator=%0A"
    for num in range(5):
        time.sleep(10)
        sss = requests.get(url)
        soup = BeautifulSoup(sss.text ,‘html.parser‘)
        guoneiip = str(soup).split("
")
        for ip in guoneiip:
            if ip:
                ipall.append(ip)
    wait_list = Queue()
    thread_num = 20        
    for keyip in ipall:
        if keyip:
            keyip = keyip.strip()#去除空格
            wait_list.put(keyip)#往Queue添加
    for i in range(thread_num):
        thread = BdSpider(wait_list)
        thread.setDaemon(True)#设置守护进程
        thread.start()
    wait_list.join()
    return allipus
allipus = []

 

传入一个

Queue   waiting当

self.waiting.task_done()此行注释程序会一直卡死,无限挂起状态。此时被认为线程没结束，被 .join()。守护。主线程结束不了

原文参考http://www.vuln.cn/8610