Python教程——20.协程 - 2
mhr18 2025-05-23 18:45 16 浏览 0 评论
异步编程
asyncio.Future 对象
Task 继承 Future, Task对象内部中的await结果的处理基于Future对象来的
在Future对象中会保存当前执行的这个协程任务的状态,如果当前任务状态为finished, 则await不再等待。
示例1:
import asyncio
async def main():
# 获取当前事件循环
loop = asyncio.get_running_loop()
# 创建一个任务[Future对象] 当前没有任何任务
fut = loop.create_future()
# 等待任务的最终结果,没有结果则一直等待
await fut
asyncio.run(main())
示例2:
import asyncio
async def set_after(fut):
await asyncio.sleep(2)
fut.set_result('这是一个测试结果')
async def main():
# 获取事件循环
loop = asyncio.get_running_loop()
# 创建一个任务, 并且当前任务没有绑定任何行为, 则这个任务永远不知道什么时候结束
fut = loop.create_future()
# 手动设置future任务的最终结果
await loop.create_task(set_after(fut))
# 等待Future对象获取最终的结果, 否则就一直等
data = await fut
print(data)
asyncio.run(main())
concurrent.futures.Future 对象
使用线程池、进程池实现异步操作时会使用到的对象。
import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor
def func(value):
time.sleep(1)
print(value)
# 创建线程池
pool = ThreadPoolExecutor(max_workers=5)
# 创建进程池
# pool = ProcessPoolExecutor(max_workers=5)
for i in range(10):
fut = pool.submit(func, i)
print(fut)
一般情况下,代码编写需要统一编程风格,简而言之,就是如果使用的是线程/进程,则整个程序都统一使用线程/进程。
只有一种情况可能会进行交叉编程。一个项目中的所有IO请求为协程异步请求,假设MySQL数据库版本过低导致无法使用协程进行并发存储,这种情况会使用线程/进程完成并发存储任务。
import time
import asyncio
import concurrent.futures
def func_1():
time.sleep(2)
return '测试'
async def main():
loop = asyncio.get_running_loop()
# 在协程函数中运行普通函数 在执行函数时,协程内部会自动创建一个线程池来运行任务
# run_in_executor()方法第一个参数为None时则默认创建一个线程池
fut = loop.run_in_executor(None, func_1)
result = await fut
print('当前方式会自动创建一个线程池去执行普通函数: ', result)
# 在协程函数中运行基于线程池的任务, 效果与以上代码一致
with concurrent.futures.ThreadPoolExecutor() as pool:
result = await loop.run_in_executor(pool, func_1)
print('在线程池中得到的执行结果: ', result)
# 在协程函数中运行基于进程池的任务
with concurrent.futures.ProcessPoolExecutor() as pool:
result = await loop.run_in_executor(pool, func_1)
print('在进程池中得到的执行结果: ', result)
if __name__ == "__main__":
asyncio.run(main())
案例:asyncio + 不支持异步的模块(requests)
import asyncio
import requests
async def download_image(url):
# 发送网络请求,下载图片(遇到网络下载图片的IO请求,自动切换到其他任务)
print('开始下载: ', url)
loop = asyncio.get_event_loop()
# requests 模块默认不支持异步操作,所以使用线程池来配合实现
future = loop.run_in_executor(None, requests.get, url)
response = await future
print('下载完成...')
# 保存图片
file_name = url.rsplit('/')[-1]
with open(file_name, mode='wb') as f:
f.write(response.content)
if __name__ == '__main__':
url_list = [
'http://pic.bizhi360.com/bbpic/98/10798.jpg',
'http://pic.bizhi360.com/bbpic/92/10792.jpg',
'http://pic.bizhi360.com/bbpic/86/10386.jpg'
]
tasks = [download_image(url) for url in url_list]
# loop = asyncio.get_event_loop()
# loop.run_until_complete(asyncio.wait(tasks))
asyncio.run(asyncio.wait(tasks))
异步迭代器
什么是异步迭代器?
实现了__aiter__() 和 __anext__() 方法的对象。__aiter__() 必须返回一个awaitable对象。async for会处理异步迭代器的 __anext__()方法所返回的可等待对象,直到引发一个StopAsyncIteration异常。
什么是异步可迭代对象?
可在async for语句中被使用的对象。必须通过它的__aiter__()方法返回一个asynchronous iterator。
import asyncio
# 自定义异步迭代器
class Reader:
def __init__(self):
self.count = 0
async def readline(self):
# await asyncio.sleep(1)
self.count += 1
if self.count == 100:
return None
return self.count
def __aiter__(self):
return self
async def __anext__(self):
val = await self.readline()
if val is None:
raise StopAsyncIteration
return val
async def func():
obj = Reader()
# 异步for循环必须在协程函数内执行,协程函数名称随意取名
async for item in obj:
print(item)
asyncio.run(func())
异步上下文管理器
此种现象通过定义__aenter__()和__axeit__()方法来对async with语句中的环境进行控制。
import asyncio
class AsyncContextManager:
def __init__(self, conn=None):
self.conn = conn
async def do_something(self):
# 异步操作数据库
return 'crud'
async def __aenter__(self):
# 异步连接数据库
self.conn = await asyncio.sleep(1)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
# 异步关闭数据库连接
await asyncio.sleep(1)
async def func():
# 上下文管理器处理也需要在协程函数中运行
async with AsyncContextManager() as f:
result = await f.do_something()
print(result)
asyncio.run(func())
uvloop
是asyncio的事件循环的替代方案。
uvloop事件循环的执行效率比asyncio默认的事件循环的效率高。
# pip install uvloop
import asyncio
import uvloop
# 设置事件循环为uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 编写的asyncio代码与之前一致
# 内部事件循环会自动切到uvloop
asyncio.run(...)
实战案例
异步操作 Redis
在使用python代码操作redis时,像连接、读取/写入、断开都是IO操作。
pip install aioredis==1.3.1
案例1:
import asyncio
import aioredis
async def execute(address):
print('开始执行: ', address)
# 网络IO 创建redis连接
redis = await aioredis.create_redis(address)
# 网络IO 在redis中设置哈希值
await redis.hmset_dict('car', key1=1, key2=2, key3=3)
# 网络IO 获取redis中的值
result = await redis.hgetall('car', encoding='utf-8')
print(result)
redis.close()
# 网络IO 关闭redis连接
await redis.wait_closed()
print('结束...')
asyncio.run(execute('redis://127.0.0.1:6379/0'))
案例2:
import asyncio
import aioredis
async def execute(address, password):
print('开始执行: ', address)
# 网络IO 创建redis连接
redis = await aioredis.create_redis_pool(address, password=password)
# 网络IO 在redis中设置哈希值
await redis.hmset_dict('car', key1=1, key2=2, key3=3)
# 网络IO 获取redis中的值
result = await redis.hgetall('car', encoding='utf-8')
print(result)
redis.close()
# 网络IO 关闭redis连接
await redis.wait_closed()
print('结束...')
task_list = [
execute('redis://localhost:6379/0', None),
execute('redis://localhost:6379/1', None)
]
asyncio.run(asyncio.wait(task_list))
异步 MySQL
pip install aiomysql
案例1:
import asyncio
import aiomysql
async def execute():
# 网络IO操作 连接mysql
conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='root', db='mysql')
# 网络IO操作 创建游标
cursor = await conn.cursor()
# 网络IO操作 执行sql
await cursor.execute('select host,user from user')
# 网络IO操作 获取sql结果
result = await cursor.fetchall()
print(result)
# 网络IO操作
await cursor.close()
conn.close()
asyncio.run(execute())
案例2:
import asyncio
import aiomysql
async def execute(host, password):
print('开始连接:', host)
# 网络IO操作 连接mysql
conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql')
# 网络IO操作 创建游标
cursor = await conn.cursor()
# 网络IO操作 执行sql
await cursor.execute('select host,user from user')
# 网络IO操作 获取sql结果
result = await cursor.fetchall()
print(result)
# 网络IO操作
await cursor.close()
conn.close()
print('结束:', host)
task_list = [
execute('localhost', 'root'),
execute('localhost', 'root')
]
asyncio.run(asyncio.wait(task_list))
FastAPI框架
pip install uvicorn
pip install fastapi
示例:
import uvicorn
import asyncio
import aioredis
from fastapi import FastAPI
app = FastAPI()
# 创建redis连接池
REDIS_POOL = aioredis.ConnectionsPool('redis://localhost:6379', password=None, minsize=1, maxsize=10)
@app.get('/')
def index():
# 普通视图函数
return {'message': 'hello world'}
@app.get('/red')
async def red():
# 异步视图
print('请求来了...')
await asyncio.sleep(3)
# 获取连接池中的一个链接
conn = await REDIS_POOL.acquire()
redis = aioredis.Redis(conn)
# 设置值
await redis.hmset_dict('car_fastApi', key1=1, key2=2, key3=3)
# 读取值
result = await redis.hgetall('car_fastApi', encoding='utf-8')
print(result)
# 将单个连接归还给连接池
REDIS_POOL.release(conn)
return result
if __name__ == '__main__':
# fastapi_test为当前这个脚本文件的名称
uvicorn.run("fastapi_test:app", host='127.0.0.1', port=5000, log_level='info')
爬虫
import asyncio
import aiohttp
async def fetch(session, url):
print('发送请求: ', url)
async with session.get(url, verify_ssl=False) as response:
text = await response.text()
print('结果: ', url, len(text))
async def main():
async with aiohttp.ClientSession() as session:
url_list = [
'https://python.org',
'https://www.baidu.com',
]
tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
await asyncio.wait(tasks)
if __name__ == '__main__':
asyncio.run(main())
相关推荐
- Dubai's AI Boom Lures Global Tech as Emirate Reinvents Itself as Middle East's Silicon Gateway
-
AI-generatedimageAsianFin--Dubaiisrapidlytransformingitselffromadesertoilhubintoaglob...
- OpenAI Releases o3-pro, Cuts o3 Prices by 80% as Deal with Google Cloud Reported to Make for Compute Needs
-
TMTPOST--OpenAIisescalatingthepricewarinlargelanguagemodel(LLM)whileseekingpartnershi...
- 黄仁勋说AI Agent才是未来!但究竟有些啥影响?
-
,抓住风口(iOS用户请用电脑端打开小程序)本期要点:详解2025年大热点你好,我是王煜全,这里是王煜全要闻评论。最近,有个词被各个科技大佬反复提及——AIAgent,智能体。黄仁勋在CES展的发布...
- 商城微服务项目组件搭建(五)——Kafka、Tomcat等安装部署
-
1、本文属于mini商城系列文档的第0章,由于篇幅原因,这篇文章拆成了6部分,本文属于第5部分2、mini商城项目详细文档及代码见CSDN:https://blog.csdn.net/Eclipse_...
- Python+Appium环境搭建与自动化教程
-
以下是保姆级教程,手把手教你搭建Python+Appium环境并实现简单的APP自动化测试:一、环境搭建(Windows系统)1.安装Python访问Python官网下载最新版(建议...
- 零配置入门:用VSCode写Java代码的正确姿
-
一、环境准备:安装JDK,让电脑“听懂”Java目标:安装Java开发工具包(JDK),配置环境变量下载JDKJava程序需要JDK(JavaDevelopmentKit)才能运行和编译。以下是两...
- Mycat的搭建以及配置与启动(mycat2)
-
1、首先开启服务器相关端口firewall-cmd--permanent--add-port=9066/tcpfirewall-cmd--permanent--add-port=80...
- kubernetes 部署mysql应用(k8s mysql部署)
-
这边仅用于测试环境,一般生产环境mysql不建议使用容器部署。这里假设安装mysql版本为mysql8.0.33一、创建MySQL配置(ConfigMap)#mysql-config.yaml...
- Spring Data Jpa 介绍和详细入门案例搭建
-
1.SpringDataJPA的概念在介绍SpringDataJPA的时候,我们首先认识下Hibernate。Hibernate是数据访问解决技术的绝对霸主,使用O/R映射(Object-Re...
- 量子点格棋上线!“天衍”邀您执子入局
-
你是否能在策略上战胜量子智能?这不仅是一场博弈更是一次量子智力的较量——量子点格棋正式上线!试试你能否赢下这场量子智局!游戏玩法详解一笔一画间的策略博弈游戏目标:封闭格子、争夺领地点格棋的基本目标是利...
- 美国将与阿联酋合作建立海外最大的人工智能数据中心
-
当地时间5月15日,美国白宫宣布与阿联酋合作建立人工智能数据中心园区,据称这是美国以外最大的人工智能园区。阿布扎比政府支持的阿联酋公司G42及多家美国公司将在阿布扎比合作建造容量为5GW的数据中心,占...
- 盘后股价大涨近8%!甲骨文的业绩及指引超预期?
-
近期,美股的AI概念股迎来了一波上升行情,微软(MSFT.US)频创新高,英伟达(NVDA.US)、台积电(TSM.US)、博通(AVGO.US)、甲骨文(ORCL.US)等多股亦出现显著上涨。而从基...
- 甲骨文预计新财年云基础设施营收将涨超70%,盘后一度涨8% | 财报见闻
-
甲骨文(Oracle)周三盘后公布财报显示,该公司第四财季业绩超预期,虽然云基建略微逊于预期,但管理层预计2026财年云基础设施营收预计将增长超过70%,同时资本支出继上年猛增三倍后,新财年将继续增至...
- Springboot数据访问(整合MongoDB)
-
SpringBoot整合MongoDB基本概念MongoDB与我们之前熟知的关系型数据库(MySQL、Oracle)不同,MongoDB是一个文档数据库,它具有所需的可伸缩性和灵活性,以及所需的查询和...
- Linux环境下,Jmeter压力测试的搭建及报错解决方法
-
概述 Jmeter最早是为了测试Tomcat的前身JServ的执行效率而诞生的。到目前为止,它的最新版本是5.3,其测试能力也不再仅仅只局限于对于Web服务器的测试,而是涵盖了数据库、JM...
你 发表评论:
欢迎- 一周热门
- 最近发表
-
- Dubai's AI Boom Lures Global Tech as Emirate Reinvents Itself as Middle East's Silicon Gateway
- OpenAI Releases o3-pro, Cuts o3 Prices by 80% as Deal with Google Cloud Reported to Make for Compute Needs
- 黄仁勋说AI Agent才是未来!但究竟有些啥影响?
- 商城微服务项目组件搭建(五)——Kafka、Tomcat等安装部署
- Python+Appium环境搭建与自动化教程
- 零配置入门:用VSCode写Java代码的正确姿
- Mycat的搭建以及配置与启动(mycat2)
- kubernetes 部署mysql应用(k8s mysql部署)
- Spring Data Jpa 介绍和详细入门案例搭建
- 量子点格棋上线!“天衍”邀您执子入局
- 标签列表
-
- oracle位图索引 (74)
- oracle批量插入数据 (65)
- oracle事务隔离级别 (59)
- oracle 空为0 (51)
- oracle主从同步 (56)
- oracle 乐观锁 (53)
- redis 命令 (78)
- php redis (88)
- redis 存储 (66)
- redis 锁 (69)
- 启动 redis (66)
- redis 时间 (56)
- redis 删除 (67)
- redis内存 (57)
- redis并发 (52)
- redis 主从 (69)
- redis 订阅 (51)
- redis 登录 (54)
- redis 面试 (58)
- 阿里 redis (59)
- redis 搭建 (53)
- redis的缓存 (55)
- lua redis (58)
- redis 连接池 (61)
- redis 限流 (51)