使用多线程队列 + urllib3 进行抓取会严重减速

Question

我试图在尽可能短的时间内抓取大量包含 JSON 格式数据的 URL（大约 300 万个）。为了实现这一点，我有一个使用队列、多线程和 Urllib3 的 Python 代码（python 3）。前 3 分钟一切正常，然后代码开始变慢，然后似乎完全卡住了。我已经阅读了我能找到的关于这个问题的所有内容，但不幸的是，解决方案似乎需要远超我的知识。

我试图限制线程数：它没有解决任何问题。我还尝试限制队列的最大大小并更改套接字超时，但它也没有帮助。远程服务器没有阻止我，也没有将我列入黑名单，因为我可以随时重新启动我的脚本，并在开始时取得良好的效果（代码在相当随机的时间开始变慢）。此外，有时我的互联网连接似乎被切断 - 因为我无法在任何网站上冲浪 - 但这个特定问题并非每次都出现。

这是我的代码（请放轻松，我是初学者）：

#!/usr/bin/env python
import urllib3,json,csv
from queue import Queue
from threading import Thread

csvFile =  open("X.csv",  'wt',newline="")
writer  =  csv.writer(csvFile,delimiter=";")
         writer.writerow(('A','B','C','D'))

def do_stuff(q):
    http = urllib3.connectionpool.connection_from_url('http://www.XXYX.com/',maxsize=30,timeout=20,block=True)

    while True:

        try:

            url = q.get()
            url1 = http.request('GET',url)      

            doc = json.loads(url1.data.decode('utf8'))

            writer.writerow((doc['A'],doc['B'], doc['C'],doc['D']))

        except:
             print(url)

        finally:
            q.task_done()

q = Queue(maxsize=200)
num_threads = 15

for i in range(num_threads):
    worker = Thread(target=do_stuff, args=(q,))
    worker.setDaemon(True)
    worker.start()

for x in range(1,3000000):
    if x < 10:
        url = "http://www.XXYX.com/?i=" + str(x) + "&plot=short&r=json"
    elif x < 100:
        url = "http://www.XXYX.com/?i=tt00000" + str(x) + "&plot=short&r=json"
    elif x < 1000:
        url = "http://www.XXYX.com/?i=0" + str(x) + "&plot=short&r=json"
    elif x < 10000:
        url = "http://www.XXYX.com/?i=00" + str(x) + "&plot=short&r=json"
    elif x < 100000:
        url = "http://www.XXYX.com/?i=000" + str(x) + "&plot=short&r=json"
    elif x < 1000000:
        url = "http://www.XXYX.com/?i=0000" + str(x) + "&plot=short&r=json"
    else:
        url = "http://www.XXYX.com/?i=00000" + str(x) + "&plot=short&r=json"

    q.put(url)

q.join()    
csvFile.close()
print("done")

Answer 1

正如 shazow 所说，这不是线程的问题，而是每个线程从服务器获取数据的超时时间。尝试在您的代码中包含一些超时：

finally:
    sleep(50)
    q.task_done()

还可以通过生成自适应超时来改进它，例如您可以测量您成功获得了多少数据，如果该数字减少，则增加睡眠时间，反之亦然