如何为两个不同的队列创建线程以并行运行？ - Python

Question

我有两个用于不同任务的队列，第一次抓取将开始抓取列表中的链接，然后它会生成更多链接以抓取到队列一，并且还会生成到队列二上不同任务的新链接，我的程序正在运行，但问题是：当队列二的工作人员开始运行时，它会阻止队列一的工作人员，他们基本上没有并行运行，他们正在等待对方完成他们的任务。我怎样才能让它们并行运行？

import threading
from queue import Queue

queue = Queue()
queue_two = Queue()

links = ['www.example.com', 'www.example.com', 'www.example.com',
         'www.example.com', 'www.example.com', 'www.example.com', 
         'www.example.com', 'www.example.com', 'www.example.com']

new_links = []

def create_workers():
    for _ in range(4):
        t = threading.Thread(target=work)
        t.daemon = True
        t.start()

    for _ in range(2):
        t = threading.Thread(target=work_two)
        t.daemon = True
        t.start()

def work():
    while True:
        work = queue.get()
        #do something
        queue.task_done()

def work_two():
    while True:
        work = queue_two.get()
        #do something
        queue_two.task_done()

def create_jobs():
    for link in links:
        queue.put(link)
    queue.join()
    crawl_two()
    crawl()

def create_jobs_two():
    for link in new_links:
        queue_two.put(link)
    queue_two.join()
    crawl_two()

def crawl():
    queued_links = links
    if len(queued_links) > 0:
        create_jobs()

def crawl_two():
    queued_links = new_links
    if len(queued_links) > 0:
        create_jobs_two()

create_workers()
crawl()

Answer 1

那是因为您的处理在工作和工作两个之间似乎并不并行。

会发生这样的事情：

1. 您为 work 和 work_two 创建工人
2. 抓取被称为
3. Create_jobs 被调用 - “工作”工作人员开始处理它们
4. Create_jobs 在queue.join() 中等待，直到所有这些都完成
5. Crawl_two 被调用
6. Create_jobs_two 被调用 - “work_two”工作人员开始处理它们
7. Create_jobs_two 在queue_two.join() 中等待，直到它们都完成
8. 调用爬取（再次从 2 开始）

基本上，您永远不会进入 work 和 work_two 并行运行的情况，因为您使用 queue.join() 等待所有当前正在运行的任务完成。只有这样，您才会将任务分配给“其他”队列。您的 work 和 work_two 确实在它们自己内部并行运行，但控制结构确保 work 和 work_two 是互斥的。如果您希望它们并行运行，则需要重新设计循环和队列。

您可能还想研究使用threading.Lock() 来保护您的全局new_links 变量，因为我假设您将在您的工作线程中附加一些东西。这绝对没问题，但是您需要一个锁来确保两个线程不会同时尝试执行此操作。但这与您当前的问题无关。这只会帮助您避免下一个问题。

我当然不知道你在这里试图实现什么，但你可以尝试通过完全废弃全局 new_links 来解决你的问题并避免下一个问题。如果您的 work 和 work_two 只是在需要向其他工作人员提交任务时将其送入队列，而不是将项目放入全局变量然后将它们送入主线程中的队列，该怎么办？

或者您可以构建一个“编排”线程，将任务排队到工作人员，处理来自他们的响应，然后相应地对该响应采取行动，要么将其排回队列之一，要么在它“准备好”时接受结果.