python - 提高Python脚本速度: Multithreading or Multiple Instances?

Question

我有一个 Python 脚本，我想每天运行它，并且我希望它只需要 1-2 小时即可运行。目前，它设置为针对给定 URL 使用 4 个不同的 API，捕获结果，然后将数据保存到 PostgreSQL 数据库中。问题是我有超过 160,000 个 URL 需要浏览，而脚本最终花费了很长的时间——我运行了一些初步测试，以当前格式浏览每个 URL 需要超过 36 个小时。所以，我的问题归结为:我应该优化我的脚本以同时运行多个线程吗？或者我应该扩大我正在使用的服务器数量？显然第二种方法成本更高，因此我更愿意在同一个实例上运行多个线程。

我正在使用我创建的库 ( SocialAnalytics )，它提供了访问不同 API 端点并解析结果的方法。以下是我配置脚本的方式:

import psycopg2
from socialanalytics import pinterest
from socialanalytics import facebook
from socialanalytics import twitter
from socialanalytics import google_plus
from time import strftime, sleep

conn = psycopg2.connect("dbname='***' user='***' host='***' password='***'")
cur = conn.cursor()

# Select all URLs
cur.execute("SELECT * FROM urls;")
urls = cur.fetchall()

for url in urls:

    # Pinterest
    try:
        p = pinterest.getPins(url[2])
    except:
        p = { 'pin_count': 0 }
    # Facebook
    try:
        f = facebook.getObject(url[2])
    except:
        f = { 'comment_count': 0, 'like_count': 0, 'share_count': 0 }
    # Twitter
    try:
        t = twitter.getShares(url[2])
    except:
        t = { 'share_count': 0 }
    # Google
    try:
        g = google_plus.getPlusOnes(url[2])
    except:
        g = { 'plus_count': 0 }

    # Save results
    try:
        now = strftime("%Y-%m-%d %H:%M:%S")
        cur.execute("INSERT INTO social_stats (fetched_at, pinterest_pins, facebook_likes, facebook_shares, facebook_comments, twitter_shares, google_plus_ones) VALUES(%s, %s, %s, %s, %s, %s, %s, %s);", (now, p['pin_count'], f['like_count'], f['share_count'], f['comment_count'], t['share_count'], g['plus_count']))
        conn.commit()
    except:
        conn.rollback()

您可以看到对 API 的每次调用都使用 Requests library ，这是一个同步、阻塞的事件。经过一些初步研究，我发现Treq ，这是一个基于 Twisted 的 API 。 Twisted 的异步、非阻塞性质似乎是改进我的方法的良好候选者，但我从未使用过它，并且我不确定它到底如何(以及是否)帮助我实现我的目标。

非常感谢任何指导!

Answer 1

首先，您应该测量脚本在每个步骤上花费的时间。也许你会发现一些有趣的东西:)

其次，您可以将网址拆分为多个 block :

chunk_size = len(urls)/cpu_core_count;//不要忘记除法的余数

完成这些步骤后，您可以使用 multiprocessing用于并行处理每个 block 。这是给您的示例:

import multiprocessing as mp

p = mp.Pool(5)

# first solution
for urls_chunk in urls: # urls = [(url1...url6),(url7...url12)...]
    res = p.map(get_social_stat, urls_chunk)
    for record in res:
        save_to_db(record)

# or, simple
res = p.map(get_social_stat, urls)

for record in res:
   save_to_db(record)

此外，gevent可以帮你。因为它可以优化同步阻塞请求的处理顺序所花费的时间。