Python中对asyncio的实际使用

news2024/11/15 13:50:41

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

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

1. 基本概念

asyncio简介

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

协程

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

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

事件循环

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

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

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

下面将分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)  # 输出所有任务的结果