关于 Python 的 asyncio 协程的问题。

先说需求：我需要处理读取多个视频，每个视频间隔一定时间取帧进行图像识别。下面是提取视频的代码，一开始我是使用多进程实现的。后面想着读取视频应该是磁盘 IO，是否使用协程合适？查资料说不是原生异步需要使用线程池维护，但是写下来，发现也没关协程啥事啊？是不是我写的有问题。正确写法应该怎么样，求各位大佬解答。

import cv2
import asyncio
import os
import numpy as np
import time

def read(cap):
    ret, frame = cap.read()
    return ret, frame

def put_frame(queue, video_path,interval=50):
    '''
    queue: 队列
    video_path: 视频路径
    interval： 间隔多少帧取一次图片
    '''
    num = 0
    img_list = []
    cap = cv2.VideoCapture(video_path)
    start = time.time()
    while cap.isOpened():
        cap.set(cv2.CAP_PROP_POS_FRAMES, num*interval)
        ret, frame = read(cap)
        if ret:
            custom_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            cv2.imshow("{0}".format(video_path), frame)
            cv2.waitKey(24)
            custom_image = cv2.resize(
            custom_image, (416, 416), interpolation=cv2.INTER_NEAREST)
            custom_image = custom_image.transpose(2, 0, 1)
            img_list.append(custom_image)
            num += 1
        else:
            arr = np.concatenate(img_list, axis=0)
            # arr = np.ascontiguousarray(arr.flat, dtype=np.float32) / 255.0
            queue.put(arr)
            print("--------------, {0}提取视频完成, 用时：{1}", video_path, (time.time()-start))
            break


video_dir = './video'
loop = asyncio.get_event_loop()
tasks = []
queue = asyncio.Queue(loop=loop, maxsize=10)
for video in os.listdir(video_dir):
    # await asyncio.sleep(0)
    video_path = os.path.join(video_dir, video)
    task = loop.run_in_executor(None, put_frame, queue, video_path)
    tasks.append(task)

loop.run_until_complete(asyncio.wait(tasks))



# 他还会出现这个警告：
C:/Users/rookie/Desktop/20200512/demo3.py:36: RuntimeWarning: coroutine 'Queue.put' was never awaited
  queue.put(arr)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

Queue

video_path

cv2

arr

16 replies • 2020-05-27 15:59:05 +08:00

yuchenyang1994

May 25, 2020

你 queue.put 没加 await await queue.push() 如果不想加，可以用 queue.put_no_wait, 还有函数没加 async 放弃吧，没啥意义这个用协程做

superlzy

May 25, 2020

await xxx.put(x)
或者
xxx.put_nowait(x)

oahebky

May 25, 2020

asyncio 在 programmer 手上一般是最上层调用。
需要较低层库支持（如此处的 CV2，OpenCV 的 asyncio 实现库）。

所以粗略来看，你这个是实现不了 asyncio 的。

======

比如 cv2.VideoCapture(video_path)
支持 asyncio 应该像 cap = await aiocv2.VideoCapture(path)
用起来长得应该就像上面这样，才是真的用 asyncio 。

superrichman

May 25, 2020 via iPhone

从你的代码来看，你这个场景是计算密集型，用异步 io 收益不大。你直接用多进程把 cpu 跑满就好。

liu826250634

May 25, 2020

@superrichman 使用协程收益确实不大，但是我想了解下协程正确写法应该是怎么样的。。。

liu826250634

May 25, 2020

@oahebky 我看着看着文档，怎么感觉 asyncio 使用起来场景限制很大，一般除了用在网络 IO，还能用在什么地方呢？
如果用网络 IO，和多线程其实也没什么区别啊，反正也是卡在网络上。

superrichman

May 25, 2020 via iPhone

@liu826250634 一处异步，处处异步。opencv2 没有做异步的支持，那处理这步就没办法异步了。

Vegetable

May 25, 2020

看到视频两个字后边都没看，视频图像这些领域，基本都是计算密集，协程说破天去也是单进程，对于计算密集的任务还是要利用好多核级的算能力。

oahebky

May 25, 2020 via Android

@liu826250634

有些语言设计之初就是异步的，所以它们的底层和它们的库在实现的时候也就是异步的；
因此对于这种语言，它们的使用场景本身就没有 Python 广泛，所以在这些语言的领域里面用起来异步是相对比较“自然”地。

不过 Python 应用场景很广泛，所以想要在不同的场景用异步，必须要不同场景的库支持才可以。
不过，即使不支持，通过消息队列把软件拆分或者其它手段也还是可以做到一整个大软件，在适合用异步的地方使用异步。

磁盘 IO 和网络 IO 都可以用异步，所以未来 Python 的异步库更加健壮之后，应该在 web server 方面可以做到大范围的异步。
比如说想象一个用户请求到达了 server，然后执行到 SQL 查询，这时候慢 IO 交出该线程控制权，就可以 handler 其它传入的 socket - 阻塞 IO 连接请求、资源请求；在 SQL 查询的零点几秒时间系统是完全有时间建立起一个新的 socket fd，或者 CPU 执行一部分运算或者发起一次新的慢 IO 的。
CPU 的执行速度要比人类想象的快得多得多。

而 CPU 密集形运算，比如处理在内存中的图像矩阵，那么在数值计算这方面就不应该玩异步上面考虑。
因此什么时候使用异步自然是要分具体应用的。

lttzzlll

May 25, 2020

如果一定要用 asyncio,需要把耗时的操作放到一个单独的进程中。

```Python
import os
import time
import asyncio
from concurrent.futures.process import ProcessPoolExecutor

pool = ProcessPoolExecutor(max_workers=os.cpu_count())

def add(a, b):
"""
耗时操作
"""
time.sleep(3)
return a + b

async def main():
loop =asyncio.get_running_loop()
r = await loop.run_in_executor(pool, add, 1, 2)
print(r)

asyncio.run(main())
```