通过线程并行化慢速 api 调用

Question

我正在运行一个 python 脚本，该脚本调用一个依赖于慢速 api 的函数，该函数又调用另一个也依赖于相同慢速 api 的函数。我想加快速度。

最好的方法是什么？线程模块？如果有，请举例说明。我注意到关于线程的一件事是，您似乎无法从线程中检索返回值……而且我的大部分脚本都是为了打印函数的返回值而编写的……

这是我试图提高 I/O 性能的代码

def get_price_eq(currency, rate):

    if isAlt(currency) == False:
        currency = currency.upper()
        price_eq = 'btc_in_usd*USD_in_'+str(currency)+'*'+str(rate)
        #print price_eq
        return price_eq 
    else:
        currency = currency.lower()
        price_eq = 'poloniex'+ str(currency) + '_close' + '*' + str(rate)
        print(price_eq)
        return price_eq

def get_btcprice_fiat(price_eq):

    query = '/api/equation/'+price_eq
    try:
        conn = api.hmac(hmac_key, hmac_secret)
        btcpricefiat = conn.call('GET', query).json()
    except requests.exceptions.RequestException as e:  # This is the correct syntax
            print(e)
    return float(btcpricefiat['data'])

usdbal = float(bal) * get_btcprice_fiat(get_price_eq('USD', 1))
egpbal = float(bal) * get_btcprice_fiat(get_price_eq('EGP', 1))
rsdbal = float(bal) * get_btcprice_fiat(get_price_eq('RSD', 1))
eurbal = float(bal) * get_btcprice_fiat(get_price_eq('EUR', 1))

如你所见，我调用了 get_btc_price，它从数据供应商处调用了一个慢速 api，并传入了另一个函数的结果，该函数使用了另一个 api 调用，我做了 4 次以上，我正在寻找增加的方法在这个功能上的表现？另外，我读过的一件事是你不能从线程中获得返回值？我的大部分代码都返回值，然后我打印给用户，我该如何处理？

Answer 1

Python 3 具有Launching parallel tasks 的功能。这使我们的工作更轻松。

它适用于thread pooling 和Process pooling。

下面给出一个见解：

ThreadPoolExecutor 示例

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
        'http://www.cnn.com/',
        'http://europe.wsj.com/',
        'http://www.bbc.co.uk/',
        'http://some-made-up-domain.com/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

ProcessPoolExecutor

import concurrent.futures
import math

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]

def is_prime(n):
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))

if __name__ == '__main__':
    main()

对于 Python 2.7，它将如下所示：

import thread
import time

# Define a function for the thread
def print_time( threadName, delay):
   count = 0
   while count < 5:
      time.sleep(delay)
      count += 1
      print "%s: %s" % ( threadName, time.ctime(time.time()) )

# Create two threads as follows
try:
   thread.start_new_thread( print_time, ("Thread-1", 2, ) )
   thread.start_new_thread( print_time, ("Thread-2", 4, ) )
except:
   print "Error: unable to start thread"

输出：

Thread-1: Thu Jan 22 15:42:17 2009
Thread-1: Thu Jan 22 15:42:19 2009
Thread-2: Thu Jan 22 15:42:19 2009
Thread-1: Thu Jan 22 15:42:21 2009
Thread-2: Thu Jan 22 15:42:23 2009
Thread-1: Thu Jan 22 15:42:23 2009
Thread-1: Thu Jan 22 15:42:25 2009
Thread-2: Thu Jan 22 15:42:27 2009
Thread-2: Thu Jan 22 15:42:31 2009
Thread-2: Thu Jan 22 15:42:35 2009

因此，在您的情况下，Python 3 将如下所示：

data = ['USD', 'EGP', 'RSD', 'EUR']
def helper_func(price_eq):
    return float(bal) * get_btcprice_fiat(get_price_eq(price_eq))


def main():
    res_dict = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        for vals, res in zip(PRIMES, executor.map(helper_func, data)):
            res_dict[vals] = res

if __name__ == '__main__':
    main()

因此，在您的情况下，Python 2.7 将如下所示：

data = ['USD', 'EGP', 'RSD', 'EUR']
final_dict = {}
def helper_func(price_eq):
    final_dict[price_eq] = float(bal) * get_btcprice_fiat(get_price_eq(price_eq))

for val in data:
    try:
        thread.start_new_thread(helper_func, (val))
    except:
        print "Error: unable to start thread for %s" % (val)