提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  其他

当前栏目

8-3-1python语法基础-并发编程-协程&asyncio&异步

amp基础并发编程异步 语法 协程 asyncio
2023-09-14 09:00:32 时间

协程

为什么要使用协程？

  • 还是为了提高性能，

如何使用异步？

  • 我要使用我自己的一个例子，
  • 上代码：
import time
import asyncio
import aiohttp

import redis
import queue
import logging

logging.basicConfig(level=logging.INFO,
                    format=
                    # 日志的时间
                    '%(asctime)s'
                    # 日志级别名称 : 当前行号
                    ' %(levelname)s [%(filename)s : %(lineno)d ]'
                    # 日志信息
                    ' : %(message)s'
                    # 指定时间格式
                    , datefmt='[%Y/%m/%d %H:%M:%S]')
logging = logging.getLogger(__name__)

# 第一步，把数据取出来，在redis
conn = redis.Redis(host="127.0.0.1", port="6379")
# proxy_list = conn.hgetall("use_proxy")
# proxy_list = conn.hvals("use_proxy")
proxy_list = conn.hkeys("use_proxy")
# logging.info(proxy_list)

# 第二步，把数据存入队列
proxy_queue = queue.Queue()
for proxy in proxy_list:
    proxy_queue.put(str(proxy, encoding="utf-8"))
queue_size = proxy_queue.qsize()


# logging.info(queue_size)
# logging.info(proxy_queue.get())


async def fetch(session):
    while True:
        try:
            proxy = proxy_queue.get(block=False)
            # print(proxy_queue.qsize())
        except queue.Empty:
            break
        headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:34.0) Gecko/20100101 Firefox/34.0',
                   'Accept': '*/*',
                   'Connection': 'keep-alive',
                   'Accept-Language': 'zh-CN,zh;q=0.8'}
        # 代理验证目标网站
        url_http = "http://httpbin.org/ip"
        url_https = "https://www.qq.com"
        http_code = False
        https_code = False
        proxy = "http://{}".format(proxy)

        # 第一个任务到这个地方遇到了阻塞，就会挂起，然后执行第二个任务，直到所有协程都执行起来，这个时候事件循环里面是有1个main协程，多个协程
        try:
            async with session.head(url_http, headers=headers, proxy=proxy, verify_ssl=False, timeout=10) as response:
                res_http = response.status
                if res_http == 200:
                    http_code = True
        except Exception as e:
            http_code = False

        try:
            async with session.get(url_https, headers=headers, proxy=proxy, verify_ssl=False, timeout=10) as response:
                res_https = response.status
                if res_https == 200:
                    https_code = True
        except Exception as e:
            https_code = False

        if http_code and https_code:
            logging.info("http_status：{} https_status：{} 代理：{} ".format(str(http_code).ljust(6),
                                                                        str(https_code).ljust(6), proxy.ljust(25)))


async def main():
    async with aiohttp.ClientSession() as session:
        task_list = []
        for i in range(20):
            # t = asyncio.create_task(fetch(session))
            # task_list.append(t)
            # 上面两行，不要这么写，这么写就是同步的效果了，就不是异步的效果了
            # # 这一步是创建task任务，并且都加入到事件循环中，里面是传入一个协程对象，
            task_list.append(asyncio.create_task(fetch(session)))

        for i in task_list:
            # 这个await 是等待返回，一直要等待返回值之后，才会往下走，这个时候main协程是挂起的状态，
            # 如果没有这个for循环，发现main函数结束了之后，hi协程并没有结束，整个的协程就结束了，
            # 所以main是主的，其他是子的，主的结束了，子的不管有没有结束都会结束，
            # 有点像是多线程里面的join，但是又不太一样，
            await i


if __name__ == '__main__':
    # 这个run，是创建事件循环，并且把main() 协程加入事件循环中，这个是一个主线程
    asyncio.run(main())

解析上面的代码

协程的实现方式

  • 在Python中有多种方式可以实现协程

greenlet

  • greenlet，是一个第三方模块，用于实现协程代码（Gevent协程就是基于greenlet实现）
  • greentlet是一个第三方模块，需要提前安装 pip3 install greenlet才能使用。
from greenlet import greenlet
def func1():
    print(1)        # 第1步：输出 1
    gr2.switch()    # 第3步：切换到 func2 函数
    print(2)        # 第6步：输出 2
    gr2.switch()    # 第7步：切换到 func2 函数，从上一次执行的位置继续向后执行
def func2():
    print(3)        # 第4步：输出 3
    gr1.switch()    # 第5步：切换到 func1 函数，从上一次执行的位置继续向后执行
    print(4)        # 第8步：输出 4
gr1 = greenlet(func1)
gr2 = greenlet(func2)
gr1.switch() # 第1步：去执行 func1 函数
注意：switch中也可以传递参数用于在切换执行时相互传递值。

yield

  • yield，生成器，借助生成器的特点也可以实现协程代码。
  • 基于Python的生成器的yield和yield form关键字实现协程代码。
def func1():
    yield 1
    yield from func2()
    yield 2
def func2():
    yield 3
    yield 4
f1 = func1()
for item in f1:
    print(item)
注意：yield form关键字是在Python3.3中引入的。

asyncio

  • asyncio，在Python3.4中引入的模块用于编写协程代码。
  • 在Python3.4之前官方未提供协程的类库，一般大家都是使用greenlet等其他来实现。
  • 在Python3.4发布后官方正式支持协程，即：asyncio模块。
import asyncio
@asyncio.coroutine
def func1():
    print(1)
    yield from asyncio.sleep(2)  # 遇到IO耗时操作，自动化切换到tasks中的其他任务
    print(2)
@asyncio.coroutine
def func2():
    print(3)
    yield from asyncio.sleep(2) # 遇到IO耗时操作，自动化切换到tasks中的其他任务
    print(4)
tasks = [
    asyncio.ensure_future( func1() ),
    asyncio.ensure_future( func2() )
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
注意：基于asyncio模块实现的协程比之前的要更厉害，因为他的内部还集成了遇到IO耗时操作自动切花的功能。

async & awiat

  • async & awiat，在Python3.5中引入的两个关键字，结合asyncio模块可以更方便的编写协程代码。
  • async & awit 关键字在Python3.5版本中正式引入，基于他编写的协程代码其实就是 上一示例 的加强版，
  • 让代码可以更加简便。
  • Python3.8之后 @asyncio.coroutine 装饰器就会被移除，推荐使用async & awit 关键字实现协程代码。

import asyncio
async def func1():
    print(1)
    await asyncio.sleep(2)
    print(2)
async def func2():
    print(3)
    await asyncio.sleep(2)
    print(4)
tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

总结

  • 关于协程有多种实现方式，
  • 基于async & await关键字的协程可以实现异步编程，这也是目前python异步相关的主流技术。
  • 目前主流使用是Python官方推荐的asyncio模块和async&await关键字的方式，
  • 例如：在tonado、sanic、fastapi、django3 中均已支持。
  • 这种方式是必须要掌握的，其他的都了解就行了，

事件循环

3.1 事件循环
事件循环，可以把他当做是一个while循环，这个while循环在周期性的运行并执行一些任务，在特定条件下终止循环。
import asyncio
loop = asyncio.get_event_loop()

协程函数

  • 函数前面加了一个async
  • 协程函数，定义形式为 async def 的函数。
  • 协程对象，调用 协程函数 所返回的对象。
# 定义一个协程函数
async def func():
    pass
# 调用协程函数，返回一个协程对象
result = func()
注意：调用协程函数时，函数内部代码不会执行，只是会返回一个协程对象。

事件循环和协程对象

  • 程序中，如果想要执行协程函数的内部代码，需要 事件循环 和 协程对象 配合才能实现，如：
import asyncio
async def func():
    print("协程内部代码")
# 调用协程函数，返回一个协程对象。
result = func()
# 方式一
# loop = asyncio.get_event_loop() # 创建一个事件循环
# loop.run_until_complete(result) # 将协程当做任务提交到事件循环的任务列表中，协程执行完成之后终止。
# 方式二
# 本质上方式一是一样的，内部先 创建事件循环 然后执行 run_until_complete，一个简便的写法。
# asyncio.run 函数在 Python 3.7 中加入 asyncio 模块，
asyncio.run(result)
这个过程可以简单理解为：将协程当做任务添加到 事件循环 的任务列表，然后事件循环检测列表中的协程是否 已准备就绪（默认可理解为就绪状态），如果准备就绪则执行其内部代码。

await

  • await是一个只能在协程函数中使用的关键字，用于遇到IO操作时挂起 当前协程（任务），
  • 当前协程（任务）挂起过程中 事件循环可以去执行其他的协程（任务），
  • 当前协程IO处理完成时，可以再次切换回来执行await之后的代码。
  • 在程序想要创建多个任务对象，需要使用Task对象来实现。

Task对象

  • Tasks用于并发调度协程，通过asyncio.create_task(协程对象)的方式创建Task对象，

  • 这样可以让协程加入事件循环中等待被调度执行。除了使用 asyncio.create_task() 函数以外，

  • 还可以用低层级的 loop.create_task() 或 ensure_future() 函数。不建议手动实例化 Task 对象。

  • 本质上是将协程对象封装成task对象，并将协程立即加入事件循环，同时追踪协程的状态。

  • 注意：

  • asyncio.create_task() 函数在 Python 3.7 中被加入。

  • 在 Python 3.7 之前，可以改用低层级的 asyncio.ensure_future() 函数。

  • 注意：

  • asyncio.wait 源码内部会对列表中的每个协程执行ensure_future从而封装为Task对象，

  • 所以在和wait配合使用时task_list的值为[func(),func()] 也是可以的。

异步上下文管理器

  • 这个异步的上下文管理器还是比较有用的，平时在开发过程中 打开、处理、关闭 操作时，就可以用这种方式来处理。
import asyncio
class AsyncContextManager:
    def __init__(self):
        self.conn = conn
    async def do_something(self):
        # 异步操作数据库
        return 666
    async def __aenter__(self):
        # 异步链接数据库
        self.conn = await asyncio.sleep(1)
        return self
    async def __aexit__(self, exc_type, 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

Python标准库中提供了asyncio模块，用于支持基于协程的异步编程。
uvloop是 asyncio 中的事件循环的替代方案，替换后可以使得asyncio性能提高。
事实上，uvloop要比nodejs、gevent等其他python异步框架至少要快2倍，性能可以比肩Go语言。
安装uvloop
pip3 install uvloop
在项目中想要使用uvloop替换asyncio的事件循环也非常简单，只要在代码中这么做就行。

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 编写asyncio的代码，与之前写的代码一致。
# 内部的事件循环自动化会变为uvloop
asyncio.run(...)
注意：知名的asgi uvicorn内部就是使用的uvloop的事件循环。

实战案例

为了更好理解，上述所有示例的IO情况都是以 asyncio.sleep 为例，而真实的项目开发中会用到很多IO的情况。

异步Redis

当通过python去操作redis时，链接、设置值、获取值 这些都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。

安装Python异步操作redis模块
pip3 install aioredis

示例1：异步操作redis。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import asyncio
import aioredis
async def execute(address, password):
    print("开始执行", address)
    # 网络IO操作：创建redis连接
    redis = await aioredis.create_redis(address, password=password)
    # 网络IO操作：在redis中设置哈希值car，内部在设三个键值对，即： redis = { car:{key1:1,key2:2,key3:3}}
    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("结束", address)
asyncio.run(execute('redis://47.93.4.198:6379', "root!2345"))
示例2：连接多个redis做操作（遇到IO会切换其他任务，提供了性能）。

import asyncio
import aioredis
async def execute(address, password):
    print("开始执行", address)
    # 网络IO操作：先去连接 47.93.4.197:6379，遇到IO则自动切换任务，去连接47.93.4.198:6379
    redis = await aioredis.create_redis_pool(address, password=password)
    # 网络IO操作：遇到IO会自动切换任务
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)
    # 网络IO操作：遇到IO会自动切换任务
    result = await redis.hgetall('car', encoding='utf-8')
    print(result)
    redis.close()
    # 网络IO操作：遇到IO会自动切换任务
    await redis.wait_closed()
    print("结束", address)
task_list = [
    execute('redis://47.93.4.197:6379', "root!2345"),
    execute('redis://47.93.4.198:6379', "root!2345")
]
asyncio.run(asyncio.wait(task_list))
更多redis操作参考aioredis官网：https://aioredis.readthedocs.io/en/v1.3.0/start.html

异步MySQL

当通过python去操作MySQL时，连接、执行SQL、关闭都涉及网络IO请求，
使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。

安装Python异步操作redis模块

pip3 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='123', db='mysql', )
    # 网络IO操作：创建CURSOR
    cur = await conn.cursor()
    # 网络IO操作：执行SQL
    await cur.execute("SELECT Host,User FROM user")
    # 网络IO操作：获取SQL结果
    result = await cur.fetchall()
    print(result)
    # 网络IO操作：关闭链接
    await cur.close()
    conn.close()
asyncio.run(execute())
示例2：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import asyncio
import aiomysql
async def execute(host, password):
    print("开始", host)
    # 网络IO操作：先去连接 47.93.40.197，遇到IO则自动切换任务，去连接47.93.40.198:6379
    conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql')
    # 网络IO操作：遇到IO会自动切换任务
    cur = await conn.cursor()
    # 网络IO操作：遇到IO会自动切换任务
    await cur.execute("SELECT Host,User FROM user")
    # 网络IO操作：遇到IO会自动切换任务
    result = await cur.fetchall()
    print(result)
    # 网络IO操作：遇到IO会自动切换任务
    await cur.close()
    conn.close()
    print("结束", host)
task_list = [
    execute('47.93.40.197', "root!2345"),
    execute('47.93.40.197', "root!2345")
]
asyncio.run(asyncio.wait(task_list))

FastAPI框架

FastAPI是一款用于构建API的高性能web框架，框架基于Python3.6+的 type hints搭建。
接下里的异步示例以FastAPI和uvicorn来讲解（uvicorn是一个支持异步的asgi）。
安装FastAPI web 框架，
pip3 install fastapi
安装uvicorn，本质上为web提供socket server的支持的asgi（一般支持异步称asgi、不支持异步称wsgi）
pip3 install uvicorn

示例：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import asyncio
import uvicorn
import aioredis
from aioredis import Redis
from fastapi import FastAPI
app = FastAPI()
REDIS_POOL = aioredis.ConnectionsPool('redis://47.193.14.198:6379', password="root123", 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 = Redis(conn)
    # 设置值
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)
    # 读取值
    result = await redis.hgetall('car', encoding='utf-8')
    print(result)
    # 连接归还连接池
    REDIS_POOL.release(conn)
    return result
if __name__ == '__main__':
    uvicorn.run("luffy:app", host="127.0.0.1", port=5000, log_level="info")
在有多个用户并发请求的情况下，异步方式来编写的接口可以在IO等待过程中去处理其他的请求，提供性能。

例如：同时有两个用户并发来向接口 http://127.0.0.1:5000/red 发送请求，服务端只有一个线程，同一时刻只有一个请求被处理。 异步处理可以提供并发是因为：当视图函数在处理第一个请求时，第二个请求此时是等待被处理的状态，当第一个请求遇到IO等待时，会自动切换去接收并处理第二个请求，当遇到IO时自动化切换至其他请求，一旦有请求IO执行完毕，则会再次回到指定请求向下继续执行其功能代码。

爬虫

在编写爬虫应用时，需要通过网络IO去请求目标数据，这种情况适合使用异步编程来提升性能，
接下来我们使用支持异步编程的aiohttp模块来实现。
安装aiohttp模块
pip3 install aiohttp

示例：

import aiohttp
import asyncio
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',
            'https://www.pythonav.com'
        ]
        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)
if __name__ == '__main__':
    asyncio.run(main())

协程总结：

  • 协程（Coroutine），也可以被称为微线程，是一种用户态内的上下文切换技术。
  • 其实就是通过一个线程实现代码块相互切换执行。
  • 但是，协程来回切换执行的意义何在呢？协程牛逼的地方到底在哪里呢？？
  • 计算型的操作，利用协程来回切换执行，没有任何意义，来回切换并保存状态 反倒会降低性能。
  • IO型的操作，利用协程在IO等待时间就去切换执行其他任务，当IO操作结束后再自动回调，
  • 那么就会大大节省资源并提供性能，从而实现异步编程（不等待任务结束就可以去执行其他代码）。

总结

  • 为了提升性能越来越多的框架都在向异步编程靠拢，目的就是用更少资源可以做处理更多的事
  • 比如：FastAPI、Tornado、Sanic、Django 3、aiohttp等。
  • 现在身边使用python的人，聊异步的也越来越多了，异步如何如何牛逼，性能如何吊炸天
  • 水涨船高，别人会，你不会你就落后了，落后就要挨打

相关文章