Python中的并发编程（4）多线程发送网络请求

多线程发送网络请求

我们使用https://www.vatcomply.com 来演示多线程发送网络请求。
该提供了汇率查询的API，我们可以像下面这样发送请求获取某种货币对其它货币的汇率。

import requests
response = requests.get("https://api.vatcomply.com/rates?base=USD")
print(response.json())
1
2
3

返回结果是一个json格式的文本，包含了base中查询的货币对其它货币的汇率：

{'date': '2023-12-07', 'base': 'USD', 'rates': {'EUR': 0.9284189026088572, 'USD': 1.0, 'JPY': 145.0004642094513, 'BGN': 1.8158016897224027, 'CZK': 22.612570791941327, ..., 'ZAR': 18.759260978553524}
}
1
2

下面我们比较不同方式发送多个请求的耗时。

注：本节代码来自Expert Python Programming 6.3

顺序执行

我们使用顺序执行的方式，发送5次请求：

import time
import requests

SYMBOLS = ('USD', 'EUR', 'PLN', 'NOK', 'CZK')
BASES = ('USD', 'EUR', 'PLN', 'NOK', 'CZK')

def fetch_rates(base):
    response = requests.get(
        f"https://api.vatcomply.com/rates?base={base}"
    )
    response.raise_for_status()
    rates = response.json()["rates"]
 
    rates[base] = 1.
    rates_line = ", ".join(
        [f"{rates[symbol]:7.03} {symbol}" for symbol in SYMBOLS]
    )
    print(f"1 {base} = {rates_line}")

def main():
    for base in BASES:
        fetch_rates(base)

if __name__ == "__main__":
    started = time.time()
    main()
    elapsed = time.time() - started
    print()
    print("time elapsed: {:.2f}s".format(elapsed))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

执行结果：

1 USD =     1.0 USD,   0.928 EUR,    4.02 PLN,    10.9 NOK,    22.6 CZK
1 EUR =    1.08 USD,     1.0 EUR,    4.33 PLN,    11.8 NOK,    24.4 CZK
1 PLN =   0.249 USD,   0.231 EUR,     1.0 PLN,    2.71 NOK,    5.62 CZK
1 NOK =  0.0916 USD,  0.0851 EUR,   0.369 PLN,     1.0 NOK,    2.07 CZK
1 CZK =  0.0442 USD,  0.0411 EUR,   0.178 PLN,   0.483 NOK,     1.0 CZK

time elapsed: 2.96s
1
2
3
4
5
6
7

顺序执行需要等待上一个请求返回后才能发起下一个请求，所以用时较长。

多线程

只需要在main函数中做一点修改，启动多个线程。

from threading import Thread
def main():
    threads = [] 
    for base in BASES:
        thread = Thread(target=fetch_rates, args=[base])
        thread.start()
        threads.append(thread)
    while threads:
        threads.pop().join()
1
2
3
4
5
6
7
8
9

执行结果：

1 PLN =   0.249 USD,   0.231 EUR,     1.0 PLN,    2.71 NOK,    5.62 CZK
1 NOK =  0.0916 USD,  0.0851 EUR,   0.369 PLN,     1.0 NOK,    2.07 CZK
1 EUR =    1.08 USD,     1.0 EUR,    4.33 PLN,    11.8 NOK,    24.4 CZK
1 USD =     1.0 USD,   0.928 EUR,    4.02 PLN,    10.9 NOK,    22.6 CZK
1 CZK =  0.0442 USD,  0.0411 EUR,   0.178 PLN,   0.483 NOK,     1.0 CZK

time elapsed: 0.62s
1
2
3
4
5
6
7

多线程的效果很好，极大地缩短了程序的耗时。因为我们连续发送了5个请求并等待结果，而不是像顺序执行中的发送一个请求后等待它返回结果后再发送下一个。
(同时我们也发现了：​多线程导致任务完成的顺序改变了， 打印的结果和启动顺序’USD’, ‘EUR’, ‘PLN’, ‘NOK’, 'CZK’不同)

但上面的代码存在一些问题：

• 没有限制线程的数量。过多的线程可能导致因请求过快而被网站封IP。
• 线程函数中使用print，可能导致输出混乱。
• 每个函数被委托给单独的线程，这使得控制输入处理的速率极其困难。

使用线程池

使用线程池创建指定数量的线程，这些线程将消耗队列中的工作项，直到队列变空。
线程池带来的好处：

• 控制线程数量
• 减少创建线程的开销。

注：这里我们用队列手动实现了线程池，但Python提供了封装好的 concurrent.futures.ThreadPoolExecutor

from queue import Empty

# 从work_queue中获取任务并执行
def worker(work_queue):
    while not work_queue.empty():
        try:
            item = work_queue.get_nowait()
        except Empty:
            break
        else:
            fetch_rates(item)
            work_queue.task_done()


from threading import Thread
from queue import Queue

THREAD_POOL_SIZE = 4
def main():
	# work_queue是任务队列
    work_queue = Queue()
    for base in BASES:
        work_queue.put(base)
        
    # 创建指定数量个线程
    threads = [
        Thread(target=worker, args=(work_queue,))
        for _ in range(THREAD_POOL_SIZE)
        ]
    for thread in threads:
        thread.start()

    work_queue.join()
    while threads:
        threads.pop().join()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

在main函数中，我们创建了一个队列work_queue来存放需要处理的参数，然后启动了指定数量THREAD_POOL_SIZE的线程。这些线程都执行worker函数，参数都是work_queue。

worker() 函数的主体是一个 while 循环，直到队列为空时结束循环。
在每次迭代中，它尝试用 work_queue.get_nowait()以非阻塞方式获取新项目。如果队列已经为空，work_queue.get_nowait()将引发 Empty 异常，从而中断循环并结束。否则从队列中获取一个项目，调用fetch_rates(item) 并用 work_queue.task_done() 将该项目标记为已处理。当队列中的所有项目都已标记为完成时，主线程中的 work_queue.join() 函数将返回。

两个队列

线程函数中使用print，有时会出现混乱的输出。
下面我们使用一个额外的队列来收集结果，并在主线程中输出结果。

首先移除原来的print函数。

def fetch_rates(base):
    response = requests.get(
        f"https://api.vatcomply.com/rates?base={base}"
    )
    response.raise_for_status()
    rates = response.json()["rates"]
 
    
    rates[base] = 1.
    # 移除print
    return base, rates

def present_result(base, rates):
    rates_line = ", ".join(
        [f"{rates[symbol]:7.03} {symbol}" for symbol in SYMBOLS]
    )
    print(f"1 {base} = {rates_line}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

修改worker函数，用results_queue收集结果：

def worker(work_queue, results_queue):
    while not work_queue.empty():
        try:
            item = work_queue.get_nowait()
        except Empty:
            break
        else:
            results_queue.put(fetch_rates(item)) # 将结果放入results_queue
            work_queue.task_done()
1
2
3
4
5
6
7
8
9

在main函数中打印结果：

def main():
    work_queue = Queue()
    results_queue = Queue()
    for base in BASES:
        work_queue.put(base)
    threads = [
        Thread(target=worker, args=(work_queue,results_queue))
        for _ in range(THREAD_POOL_SIZE)
        ]
    for thread in threads:
        thread.start()

    work_queue.join()
    while threads:
        threads.pop().join()
    
    # 打印结果
    while not results_queue.empty():
        present_result(*results_queue.get())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

处理线程中的错误

我们的fetch_rates函数向网站发送请求时可能因为网络等原因出错，然后该线程会结束（但该任务没有完成）。主线程中的work_queue.join()会等待所有任务完成，从而程序被卡住。

我们通过在fetch_rates中添加一个随机报错模拟网络出错的情况：

import random
def fetch_rates(base):
    response = requests.get(
        f"https://api.vatcomply.com/rates?base={base}"
    )
    # 随机引起一个报错
    if random.randint(0, 5) < 1:
        # simulate error by overriding status code
        response.status_code = 500
    response.raise_for_status()
    rates = response.json()["rates"]
    

    # note: same currency exchanges to itself 1:1
    rates[base] = 1.
    return base, rates
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

如果出现了错误（异常），程序将抛出异常，然后卡住。

因此我们需要在worker中添加异常处理。
当发生异常时，程序将异常存入results_queue中；如果没有异常，则存放正常的结果；并且总是该标记任务完成。

def worker(work_queue, results_queue):
    while not work_queue.empty():
        try:
            item = work_queue.get_nowait()
        except Empty:
            break
        # 处理错误
        try:
            result = fetch_rates(item)
        except Exception as err:
            results_queue.put(err)
        else:
            results_queue.put(result)
        finally:
            work_queue.task_done()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

在main函数中：

    # 打印结果
    while not results_queue.empty():
        result = results_queue.get()
        if isinstance(result, Exception):
            raise result
        present_result(*result)
1
2
3
4
5
6

程序遇到错误时，不会再卡住，在最后的打印时会抛出(raise)错误。

Throttling（节流）

过快的请求可能导致网站负载过大，从而封禁我们的IP。
因此我们需要控制请求的速度。
我们将使用的算法有时称为令牌桶(token bucket)，非常简单。它包括以下功能：
• 有一个包含预定义数量令牌的存储桶
• 每个令牌对应于处理一项工作的单个权限
• 每次工作人员请求一个或多个令牌（权限）时，我们都会执行以下操作：
1. 我们检查自上次重新装满桶以来已经过去了多长时间
2. 如果时间差允许，我们将与时间差相对应的令牌数量重新装满桶
3. 如果存储的数量令牌大于或等于请求的数量，我们减少存储的令牌数量并返回该值
4. 如果存储的令牌数量小于请求的数量，我们返回零

两件重要的事情是
1.始终用零令牌初始化令牌桶（？）
2.并且永远不要让它溢出。

from threading import Lock 
import time

class Throttle: 
    def __init__(self, rate): 
        self._consume_lock = Lock() # 使用锁避免冲突
        self.rate = rate # 速率，rate越大则允许的请求间隔越小
        self.tokens = 0
        self.last = None

    def consume(self, amount=1): 
        with self._consume_lock: 
            now = time.time() 
            #初始化上次时间
            if self.last is None: 
                self.last = now 
            elapsed = now - self.last 
            # 间隔时间足够，增加令牌
            if elapsed * self.rate > 1: 
                self.tokens += elapsed * self.rate
                self.last = now 
            # 避免桶溢出
            self.tokens = min(self.rate, self.tokens)
            # 如果令牌足够，则发给请求的进程
            if self.tokens >= amount: 
                self.tokens -= amount
                return amount

            return 0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

这个类的用法非常简单。
我们只需在主线程中创建一个 Throttle 实例（例如 Throttle(10)，rate=10，允许每1/10秒发送一个请求，rate越大则允许的请求速度越快），并将其作为参数传递给每个工作线程：

	throttle = Throttle(10)
	...
	threads = [
        Thread(target=worker, 
               args=(work_queue, results_queue, throttle)
               )
        for _ in range(THREAD_POOL_SIZE)
        ]
1
2
3
4
5
6
7
8

在worker中，需要消耗throttle：

def worker(work_queue, results_queue, throttle):
    while not work_queue.empty():
        try:
            item = work_queue.get_nowait()
        except Empty:
            break
            
		# 尝试获取和消耗令牌
        while not throttle.consume():
            time.sleep(.1)

        # 处理错误
        ...
1
2
3
4
5
6
7
8
9
10
11
12
13

如果没有足够的令牌，则该线程会等待直到能够获取令牌时继续执行。