Python如何批量处理多线程？

Question

我正在用 Python 构建一个简单的网络爬虫。我将不得不浏览大约 50k 个网站，我想通过一些多线程来加快这个过程。

我已经定义了一个爬虫类来爬取每个网站作为线程的元对象:

Crawler(Thread):
     def __init__(self, url, depth, wait):
...

然后在主函数中，我从完整的 URL 列表中逐批迭代 10 个 URL，并为每个 URL 创建一个 Crawler 对象:

    for i in range(index, math.ceil(len(urls) / 10)):
        jobs = []
        for url in urls[i * 10:(i + 1) * 10]:
            s = Crawler(url)
            s.setDaemon(True)
            s.start()
            jobs.append(s)

        for j in jobs:
            j.join()

问题是，对于每个批处理，我都必须等待所有线程完成。这是低效的，因为当我有 9 个 100 页的网站和只有 1 个 10,000 页的网站时，这 9 个网站将在几分钟内完成，但我将不得不等待一个小时才能让 10,000 页的大型网站完成在我继续下一批之前完成。

为了优化，最好从 10 个 Crawler 线程开始，然后，每次 Crawler 线程完成时，使用列表中的下一个 url 创建一个新的 Crawler，直到列表完成。

我想我可以摆脱 join() 并在 threading.enumerate 的长度上有一个 while 循环，每当长度低于 10 时添加一个新线程，但这听起来有点骇人听闻。

我正在研究 python 的 Queue 但从 https://docs.python.org/3/library/queue.html 的例子来看，我仍然必须依赖 .join() 并因此等待队列中的所有线程都已执行。

有什么方法可以向线程添加类似“事件监听器”的东西，以便每当线程结束时我都可以用新线程更新线程列表？

Answer 1

也许再看看队列，您不需要每批或根本不需要加入。

您可以将所有 50K 网站放入队列中。我可能会调用 jobs 并且有限数量的线程通常称为 workers。然后每个工作人员从队列中挑选一个项目，处理它并继续从队列中挑选项目直到完成。完成的意思各不相同。一个建议是将 None 放入每个 worker 的队列中，并且每个 worker 都会在看到 None 时停止。但是您还可以使用其他信号。然后您可以使用 join 等待所有工作线程完成。在那种情况下， worker 不需要是守护进程。 (您不想为每个 URL 创建一个单独的线程)

例如:

from threading import Queue, Thread

def crawl_worker(q):
  while True:
    url = q.get()
    if url is None:
      break
    # do something with url

url_queue = Queue()

# populate the queue
for url in urls:
  url_queue.put(url)

num_workers = 10

workers = [
  Thread(target=crawl_worker, args=(url_queue))
  for _ in range(num_workers)
]

# add a None signal for each worker
for worker in workers:
  url_queue.put(None)

# start all workers
for worker in workers:
  worker.start()

# wait for all workers to finish
for worker in workers:
  worker.join()

# we're done now

有替代方案。如果您实际上不是将其视为练习，而是想完成某件事，那么我建议 https://scrapy.org/