Python中对asyncio的实际使用

发布于:2024-07-11 ⋅ 阅读:(25) ⋅ 点赞:(0)

前言:一般涉及异步编程我都无脑用celery,但是最近在做一个项目,项目不大,也不涉及定时任务,所以就用了asyncio。

asyncio是python自带的模块,比celery轻量,使用起来也简单。以前学习过,但是公司项目中碰到并发任务基本都用celery处理,所以没有使用,这次就简单总结一下。

1. 基本概念

asyncio简介

asyncio是Python的标准库,它提供了一种异步编程的框架,可以用于编写并发程序。asyncio使用事件循环来管理异步任务,通过协程(coroutine)来实现异步操作。

协程

协程又叫微线程,是一种特殊的函数,它可以挂起和恢复执行,而不阻塞事件循环。协程使用async和await关键字来定义和调用。

通俗的理解:在一个线程中的某个函数,可以在任何地方保存当前函数的一些临时变量等信息,然后切换到另外一个函数中执行另一个函数。 简而言之,其实就是通过一个线程实现代码块相互切换执行。

事件循环

事件循环是一个持续运行的循环,用于调度和执行任务。它不断地检查是否有任务需要执行,并在任务完成时继续处理其他任务。

2.实际项目场景--开发盘点系统

仓库有多台堆垛机,输入盘点任务,盘点系统需要根据盘点货位号通知堆垛机到达指定位置,堆垛机到达指定货位后通知盘点系统同时进行RFID扫描和视觉比对。

image.png

下面将分2个部分介绍asyncio的使用,第一部分是使用asyncio实现并发网络请求,如图中①所示,第二部分是使用asyncio实现异步处理长时间的耗时任务,如图中②所示

3.asyncio实现并发网络请求

asyncio实现并发网络请求,需要使用asyncio的异步网络请求库aiohttp。aiohttp是一个基于asyncio的异步HTTP客户端/服务器库,可以用于发送HTTP请求和创建HTTP服务器。

示例如下,发送请求通知堆垛机到达指定货位,请求地址相同,但是请求参数(货位号)不同,使用asyncio实现并发请求,节省盘点时间。

import asyncio
import datetime
import aiohttp

async def async_post(url, data,headers):
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=data,headers=headers) as response:
            return await response.json()


async def execute_queue(payloads):
    url = 'http://127.0.0.1:8011/api/smart_check/execute_queue/'
    headers = {"content-type": "application/json"}
    tasks = []
    for payload in payloads:
        task = asyncio.create_task(async_post(url, payload,headers))
        tasks.append(task)
    responses = await asyncio.gather(*tasks)

    for response in responses:
    #打印响应结果和时间
        print(response,datetime.datetime.now())

# Run the asyncio event loop
if __name__ == "__main__":
    payloads = [
        {'space_number': '1号'}, # 货位号
        {'space_number': '8号'}, # 货位号
        {'space_number': '10号'} # 货位号
    ]
    #loop = asyncio.get_event_loop() 
    #loop.run_until_complete(result) 
    asyncio.run(execute_queue(payloads))# python3.7后支持,等同于上面两行代码
    
    
"""==========返回的结果==========="""
{'code': 200, 'data': None, 'message': '堆垛机已到达货位1号'} 2024-07-09 15:47:06.778444
{'code': 200, 'data': None, 'message': '堆垛机已到达货位8号'} 2024-07-09 15:47:06.778444
{'code': 200, 'data': None, 'message': '堆垛机已到达货位10号'} 2024-07-09 15:47:06.778444
"""=============================="""

4.asyncio实现异步处理长时间的耗时任务

4.1 使用asyncios实现

很多时候耗时任务根本没有aiohttp这样的模块帮我们实现,所以可以用 run_in_executor 将耗时任务放入一个线程池执行器中运行,以避免阻塞事件循环所在的线程。简单理解就是调用新的线程去执行异步任务。demo如下:

import time
import asyncio

def task():
    #某个耗时操作任务
    time.sleep(2)
    return "ok"

async def main():
    loop = asyncio.get_event_loop()
    # 如果第一个参数不填,默认会使用一个 `ThreadPoolExecutor`,它会在一个线程池中执行函数,并返回返回一个Future对象,而Future就是一个可等待对象
    res = loop.run_in_executor(None,task)
    result = await res
    print("default thread pool",result)

asyncio.run(main())

实际开发中:

import time
import asyncio

def rfid():
    time.sleep(2)
    return "读取RFID信息"

def capture():
    time.sleep(6)
    return "摄像头抓图"

async def main():
    loop = asyncio.get_event_loop()
    tasks = [loop.run_in_executor(None,rfid),loop.run_in_executor(None,capture)]
    res = await asyncio.gather(*tasks)
    print(res)
    # 判断有没有待盘点的货位,如果有就继续通知堆垛机,此处省略---
    # payloads = [
    #   {'space_number': '1号'}, # 货位号
    #]
    #asyncio.run(execute_queue(payloads))

if __name__ == "__main__":
    asyncio.run(main())

最后打印的结果是['读取RFID信息', '摄像头抓图']。asyncio会保证两个异步任务都结束后才会继续执行execute_queue函数,就是通知堆垛机处理剩余待盘点的货位。

4.2 使用celery实现

如果用celery则是需要用group做处理,才能保证这两个异步任务都结束,而且是并发执行,因为会同时分配给两个worker去执行。例子如下:

from celery import group

# 定义多个任务
@celery.task
def rfid():
    time.sleep(2)
    return "读取RFID信息"

@celery.task
def capture():
    time.sleep(6)
    return "摄像头抓图"
    
# 将任务组合成一个 group
job_group = group(rfid.s(), capture.s())

# 执行 group 中的任务
result = job_group.apply_async()

# 等待所有任务完成并获取结果
final_result = result.get()

print(final_result)  # 输出所有任务的结果
wAAACH5BAEKAAAALAAAAAABAAEAAAICRAEAOw==

5. 总结

celery和asyncio写代码都差不多,但asycio用起来更简单,更适用于网络并发请求。如果用于做耗时任务处理也可以,针对如果耗时任务只有一个,明显用celery把耗时任务转到后台处理更为合适。

作者:大海前端
链接:https://juejin.cn/post/7389912354747990027
来源:稀土掘金
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。