Python中的并发编程(4)多线程发送网络请求

简介: Python中的并发编程(4)多线程发送网络请求

多线程发送网络请求

我们使用https://www.vatcomply.com 来演示多线程发送网络请求。

该提供了汇率查询的API,我们可以像下面这样发送请求获取某种货币对其它货币的汇率。

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

返回结果是一个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}
}

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

注:本节代码来自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 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

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

多线程

只需要在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 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

多线程的效果很好,极大地缩短了程序的耗时。因为我们连续发送了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()

在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}")

修改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()

在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())

处理线程中的错误

我们的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

如果出现了错误(异常),程序将抛出异常,然后卡住。

因此我们需要在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()

在main函数中:

    # 打印结果
    while not results_queue.empty():
        result = results_queue.get()
        if isinstance(result, Exception):
            raise result
        present_result(*result)


程序遇到错误时,不会再卡住,在最后的打印时会抛出(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

这个类的用法非常简单。

我们只需在主线程中创建一个 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)
        ]

在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)

        # 处理错误
        ...
相关文章
|
4天前
|
并行计算 数据处理 Python
Python并发编程迷雾:IO密集型为何偏爱异步?CPU密集型又该如何应对?
【7月更文挑战第17天】Python并发编程中,异步编程(如`asyncio`)在IO密集型任务中提高效率,利用等待时间执行其他任务。但对CPU密集型任务,由于GIL限制,多线程效率不高,此时应选用`multiprocessing`进行多进程并行计算以突破限制。选择合适的并发策略是关键:异步适合IO,多进程适合CPU。理解这些能帮助构建高效并发程序。
15 6
|
5天前
|
安全 Python
在Python中,实现多线程
【7月更文挑战第16天】在Python中,实现多线程
17 6
|
3天前
|
开发框架 并行计算 .NET
从菜鸟到大神:Python并发编程深度剖析,IO与CPU的异步战争!
【7月更文挑战第18天】Python并发涉及多线程、多进程和异步IO(asyncio)。异步IO适合IO密集型任务,如并发HTTP请求,能避免等待提高效率。多进程在CPU密集型任务中更优,因可绕过GIL限制实现并行计算。通过正确选择并发策略,开发者能提升应用性能和响应速度。
|
6天前
|
数据处理 Python
深入探索:Python中的并发编程新纪元——协程与异步函数解析
【7月更文挑战第15天】Python 3.5+引入的协程和异步函数革新了并发编程。协程,轻量级线程,由程序控制切换,降低开销。异步函数是协程的高级形式,允许等待异步操作。通过`asyncio`库,如示例所示,能并发执行任务,提高I/O密集型任务效率,实现并发而非并行,优化CPU利用率。理解和掌握这些工具对于构建高效网络应用至关重要。
20 6
|
3天前
|
UED 开发者 Python
Python并发编程新纪元:异步编程如何重塑IO与CPU密集型任务的处理方式?
【7月更文挑战第18天】Python异步编程提升IO任务效率,非阻塞模式减少等待时间,优化用户体验。asyncio库与await关键字助力编写非阻塞代码,示例展示异步HTTP请求。CPU密集型任务中,异步编程结合多进程可提升效率。异步编程挑战包括代码复杂性,解决策略包括使用类型提示、异步框架及最佳实践。异步编程重塑任务处理方式,成为现代Python开发的关键。
8 2
|
5天前
|
数据采集 并行计算 数据处理
工具人必看:Python并发编程工具箱大揭秘,IO与CPU密集型任务的最佳拍档!
【7月更文挑战第16天】Python并发编程助力IO密集型(asyncio+aiohttp,异步Web爬虫示例)和CPU密集型(multiprocessing,并行计算数组和)任务。asyncio利用单线程异步IO提升Web应用效率,multiprocessing通过多进程克服GIL限制,实现多核并行计算。善用这些工具,可优化不同场景下的程序性能。
10 1
|
6天前
|
调度 Python
揭秘Python并发编程核心:深入理解协程与异步函数的工作原理
【7月更文挑战第15天】Python异步编程借助协程和async/await提升并发性能,减少资源消耗。协程(async def)轻量级、用户态,便于控制。事件循环,如`asyncio.get_event_loop()`,调度任务执行。异步函数内的await关键词用于协程间切换。回调和Future对象简化异步结果处理。理解这些概念能写出高效、易维护的异步代码。
12 2
|
7天前
|
消息中间件 安全 数据处理
Python中的并发编程:理解多线程与多进程的区别与应用
在Python编程中,理解并发编程是提高程序性能和响应速度的关键。本文将深入探讨多线程和多进程的区别、适用场景及实际应用,帮助开发者更好地利用Python进行并发编程。
|
6天前
|
JSON 数据挖掘 API
在会议系统工程中,Python可以用于多种任务,如网络请求(用于视频会议的连接和会议数据的传输)、数据分析(用于分析会议参与者的行为或会议效果)等。
在会议系统工程中,Python可以用于多种任务,如网络请求(用于视频会议的连接和会议数据的传输)、数据分析(用于分析会议参与者的行为或会议效果)等。
|
12天前
|
安全 数据安全/隐私保护 数据中心
Python并发编程大挑战:线程安全VS进程隔离,你的选择影响深远!
【7月更文挑战第9天】Python并发:线程共享内存,高效但需处理线程安全(GIL限制并发),适合IO密集型;进程独立内存,安全但通信复杂,适合CPU密集型。使用`threading.Lock`保证线程安全,`multiprocessing.Queue`实现进程间通信。选择取决于任务性质和性能需求。
23 1