Python 线程/队列问题

Question

我正在创建一个线程Python脚本，其中有一组文件被放入队列中，然后未知数量的线程(默认为3)开始下载。当每个线程完成时，它会使用队列状态和百分比更新标准输出。所有文件都正在下载，但第三个线程的状态信息是错误的，我不知道为什么。我一直在考虑创建一个 work_completed 队列用于计算，但认为我不应该这样做/这很重要。有人能指出我正确的方向吗？

download_queue = queue.Queue()

class Downloader(threading.Thread):
    def __init__(self,work_queue):
        super().__init__()
        self.current_job = 0
        self.work_queue = work_queue
        self.queue_size = work_queue.qsize()

    def run(self):
        while self.work_queue.qsize() > 0:
            url = self.work_queue.get(True)
            system_call = "wget -nc -q {0} -O {1}".format(url,local_file)
            os.system(system_call)
            self.current_job = int(self.queue_size) - int(self.work_queue.qsize())
            self.percent = (self.current_job / self.queue_size) * 100
            sys.stdout.flush()
            status = "\rDownloading " + url.split('/')[-1] + " [status: " + str(self.current_job) + "/" + str(self.queue_size) + ", " + str(round(self.percent,2)) + "%]"
        finally:
            self.work_queue.task_done()
def main:
    if download_queue.qsize() > 0:
        if options.active_downloads:
            active_downloads = options.active_downloads
        else:
            active_downloads = 3
        for x in range(active_downloads):
            downloader = Downloader(download_queue)
            downloader.start()
        download_queue.join()

Answer 1

您无法在一条语句中检查队列大小，然后在下一条语句中检查队列中的.get()。与此同时，整个世界可能已经改变。 .get() 方法调用是您需要调用的单个原子操作。如果它引发 Empty 或阻塞，则队列为空。

您的线程可以覆盖彼此的输出。我将有另一个带有输入队列的线程，其唯一的工作是将队列中的项目打印到标准输出。它还可以计算已完成项目的数量并生成状态信息。

我也倾向于不子类化 Thread，而是只提供一个带有 target= 参数和 的普通 Thread 实例。 start() 线程。

根据您的回复，尝试以下操作:

download_queue = queue.Queue()


class Downloader(threading.Thread):
    def __init__(self,work_queue, original_size):
        super().__init__()
        self.current_job = 0
        self.work_queue = work_queue
        self.queue_size = original_size

    def run(self):
        while True:
            try:
                url = self.work_queue.get(False)
                system_call = "wget -nc -q {0} -O {1}".format(url,local_file)
                os.system(system_call)
                # the following code is questionable. By the time we get here,
                #   many other items may have been taken off the queue. 
                self.current_job = int(self.queue_size) - int(self.work_queue.qsize())
                self.percent = (self.current_job / self.queue_size) * 100
                sys.stdout.flush()
                status = ("\rDownloading " + url.split('/')[-1] + 
                          " [status: " + str(self.current_job) + 
                          "/" + str(self.queue_size) + ", " + 
                          str(round(self.percent,2)) + "%]" )            
            except queue.Empty:
                pass
            finally: 
                self.work_queue.task_done()




def main:
    if download_queue.qsize() > 0:
        original_size = download_queue.qsize()
        if options.active_downloads:
            active_downloads = options.active_downloads
        else:
            active_downloads = 3
        for x in range(active_downloads):
            downloader = Downloader(download_queue, original_size)
            downloader.start()
        download_queue.join()