百度360必应搜狗淘宝本站头条
当前位置:网站首页 > IT技术 > 正文

Python教程——20.协程 - 2 python2.7 协程

wptr33 2024-12-25 16:03 51 浏览

异步编程

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

相关推荐

[常用工具] git基础学习笔记_git工具有哪些

添加推送信息,-m=messagegitcommit-m“添加注释”查看状态...

centos7安装部署gitlab_centos7安装git服务器

一、Gitlab介1.1gitlab信息GitLab是利用RubyonRails一个开源的版本管理系统,实现一个自托管的Git项目仓库,可通过Web界面进行访问公开的或者私人项目。...

太高效了!玩了这么久的Linux,居然不知道这7个终端快捷键

作为Linux用户,大家肯定在Linux终端下敲过无数的命令。有的命令很短,比如:ls、cd、pwd之类,这种命令大家毫无压力。但是,有些命令就比较长了,比如:...

提高开发速度还能保证质量的10个小窍门

养成坏习惯真是分分钟的事儿,而养成好习惯却很难。我发现,把那些对我有用的习惯写下来,能让我坚持住已经花心思养成的好习惯。...

版本管理最好用的工具,你懂多少?

版本控制(Revisioncontrol)是一种在开发的过程中用于管理我们对文件、目录或工程等内容的修改历史,方便查看更改历史记录,备份以便恢复以前的版本的软件工程技术。...

Git回退到某个版本_git回退到某个版本详细步骤

在开发过程,有时会遇到合并代码或者合并主分支代码导致自己分支代码冲突等问题,这时我们需要回退到某个commit_id版本1,查看所有历史版本,获取git的某个历史版本id...

Kubernetes + Jenkins + Harbor 全景实战手册

Kubernetes+Jenkins+Harbor全景实战手册在现代企业级DevOps体系中,Kubernetes(K8s)、Jenkins和Harbor组成的CI/CD流水...

git常用命令整理_git常见命令

一、Git仓库完整迁移完整迁移,就是指,不仅将所有代码移植到新的仓库,而且要保留所有的commit记录1.随便找个文件夹,从原地址克隆一份裸版本库...

第三章:Git分支管理(多人协作基础)

3.1分支基本概念分支是Git最强大的功能之一,它允许你在主线之外创建独立的开发线路,互不干扰。理解分支的工作原理是掌握Git的关键。核心概念:HEAD:指向当前分支的指针...

云效Codeup怎么创建分支并进行分支管理

云效Codeup怎么创建分支并进行分支管理,分支是为了将修改记录分叉备份保存,不受其他分支的影响,所以在同一个代码库里可以同时进行多个修改。创建仓库时,会自动创建Master分支作为默认分支,后续...

git 如何删除本地和远程分支?_git怎么删除远程仓库

Git分支对于开发人员来说是一项强大的功能,但要维护干净的存储库,就需要知道如何删除过时的分支。本指南涵盖了您需要了解的有关本地和远程删除Git分支的所有信息。了解Git分支...

git 实现一份代码push到两个git地址上

一直以来想把自己的博客代码托管到github和coding上想一次更改一次push两个地址一起更新今天有空查资料实践了下本博客的github地址coding的git地址如果是Gi...

git操作:cherry-pick和rebase_git cherry-pick bad object

在编码中经常涉及到分支之间的代码同步问题,那就需要cherry-pick和rebase命令问题:如何将某个分支的多个commit合并到另一个分支,并在另一个分支只保留一个commit记录解答:假设有两...

模型文件硬塞进 Git,GitHub 直接打回原形:使用Git-LFS管理大文件

前言最近接手了一个计算机视觉项目代码是屎山就不说了,反正我也不看代码主要就是构建一下docker镜像,测试一下部署的兼容性这本来不难但是,国内服务器的网络环境实在是恶劣,需要配置各种镜像(dock...

防弹少年团田柾国《Euphoria》2周年 获世界实时趋势榜1位 恭喜呀

当天韩国时间凌晨3时左右,该曲在Twitter上以“2YearsWithEuphoria”的HashTag登上了世界趋势1位。在韩国推特实时趋势中,从上午开始到现在“Euphoria2岁”的Has...