标签: python-asyncio

我遇到了一个奇怪的问题asyncio.Queue- 不是在项目可用时立即返回，而是等待队列已满，然后再返回任何内容。我意识到在使用队列来存储从 收集的帧时cv2.VideoCapture，maxsize队列越大，在屏幕上显示任何内容所需的时间就越长，然后，它看起来像是收集到队列中的所有帧的序列。
这是一个功能，一个错误，还是我只是用错了？
无论如何，这是我的代码

import asyncio  
import cv2  
import numpy as np  


async def collecting_loop(queue):  
    print("cl")  
    cap = cv2.VideoCapture(0)  
    while True:  
        _, img = cap.read()  
        await queue.put(img)  


async def processing_loop(queue):  
    print("pl")  
    await asyncio.sleep(0.1)  
    while True:  
        img = await queue.get()  
        cv2.imshow('img', img)  
        cv2.waitKey(5)  


async def main(e_loop):  
    print("running main")  
    queue = asyncio.Queue(loop=e_loop, maxsize=10)
    await asyncio.gather(collecting_loop(queue), processing_loop(queue))


loop = asyncio.get_event_loop()   
try:   
    loop.run_until_complete(main(e_loop=loop))   
except KeyboardInterrupt:   
    pass   
finally:   
    loop.close()   

我正在尝试为 asyncio 事件循环创建一个周期性任务，如下所示，但是我收到了“RuntimeError：无法重用已经等待的协程”异常。显然， asyncio 不允许等待与此错误线程中讨论的相同的可等待函数。这就是我尝试实现它的方式：

import asyncio    

class AsyncEventLoop:    

    def __init__(self):
        self._loop = asyncio.get_event_loop()

    def add_periodic_task(self, async_func, interval):
        async def wrapper(_async_func, _interval):
            while True:
                await _async_func               # This is where it goes wrong
                await asyncio.sleep(_interval)
        self._loop.create_task(wrapper(async_func, interval))
        return

    def start(self):
        self._loop.run_forever()
        return

由于我的 while 循环，相同的可等待函数 (_async_func) 将在中间有一个睡眠间隔的情况下执行。我从如何使用 asyncio 定期执行函数中获得了实现周期性任务的灵感？ .

从上面提到的错误线程，我推断 RuntimeError 背后的想法是为了让开发人员不会意外地等待同一个协程两次或更多次，因为协程将被标记为完成并产生 None 而不是结果。有没有办法可以不止一次等待相同的功能？

我刚刚阅读了这个介绍，但是在实现其中一个示例时遇到了麻烦（注释代码是第二个示例）：

import asyncio
import pandas as pd
from openpyxl import load_workbook

async def loop_dfs(dfs):
    async def clean_df(df):
        df.drop(["column_1"], axis=1, inplace=True)
        ... a bunch of other inplace=True functions ...
        return "Done"

    # tasks = [clean_df(df) for (table, dfs) in dfs.items()]
    # await asyncio.gather(*tasks)

    tasks = [clean_df(df) for (table, df) in dfs.items()]
    completed, pending = await asyncio.wait(tasks)


def main():
    dfs = {
        sn: pd.read_excel("excel.xlsx", sheet_name=sn)
        for sn in load_workbook("excel.xlsx").sheetnames
    }

    # loop = asyncio.get_event_loop()
    # loop.run_until_complete(loop_dfs(dfs))

    loop = asyncio.get_event_loop()
    try: …

我正在编写一个包含 20,000 个任务的脚本，每个任务都会进行一个子进程调用和一个或两个 TCP 调用。为了使这不会花费一整天，我正在使用 Python 的新asyncio.

但是，我担心脚本运行时 Python 打印出的这些错误：

Exception ignored when trying to write to the signal wakeup fd:
BlockingIOError: [Errno 11] Resource temporarily unavailable

它将打印一堆，但不会引发任何异常。我之前确实OSError了解Too many open files并断开了与服务器的连接，但是我使用信号量一次只允许与每个服务器建立 100 个连接，总共只有 700 个连接。

由于 Python 没有引发任何异常，因此我无法捕获错误。不过，这并不似乎影响脚本。

这些错误是我应该关注的吗？ 如果是这样，我需要做什么才能摆脱它们？如果没有，我如何摆脱它们，使它们不在我的程序输出中？

另外，如果这些错误很严重，为什么 Python 会忽略它们而不是引发异常？

我有一个 asyncio 应用程序，它使用服务器 fromaiohttp和异步套接字asyncio.open_connection()

我的代码包含来自 PIL 库的一些阻塞调用，例如

Image.save()
Image.resize()

1. 即使调用不会阻塞太长时间，如果我使用这些阻塞调用，我的 Web 服务器是否会冻结？更准确地说，事件循环是否有可能因为阻塞代码而错过事件？
2. 如果是，与 asyncio 集成的这些功能的替代品是什么？没有异步版本的 PIL。
3. 一般来说，什么被认为是 asyncio 中的“阻塞代码”？除了像socket，读文件等明显的操作。
   例如，是否os.path.join()被认为可以？在numpy数组上工作怎么样？

我正在尝试使用 aiohttp 进行异步处理，运行具有多个工作人员的 gunicorn 服务器。

当使用一个工作进程运行服务时一切正常，但是当使用多个进程时服务失败

似乎是因为错误：https : //bugs.python.org/issue22087 是否有解决此错误的方法？

最小代码：

from aiohttp import web


async def handle(request):
    return web.Response(text='')

app = web.Application()
app.add_routes([web.get('/', handle)])

web.run_app(app)

命令： gunicorn aio.simple_server:app --worker-class aiohttp.GunicornWebWorker --workers 2

错误：

$ gunicorn aio.simple_server:app  --worker-class aiohttp.GunicornWebWorker --workers 2
[2019-02-22 23:26:32 +0200] [30784] [INFO] Starting gunicorn 19.9.0
[2019-02-22 23:26:32 +0200] [30784] [INFO] Listening at: http://127.0.0.1:8000 (30784)
[2019-02-22 23:26:32 +0200] [30784] [INFO] Using worker: aiohttp.GunicornWebWorker
[2019-02-22 23:26:32 +0200] [30787] [INFO] Booting worker with pid: 30787
======== Running on …

我试图了解如何使用双向流处理 grpc api（使用 Python API）。

假设我有以下简单的服务器定义：

syntax = "proto3";
package simple;

service TestService {
  rpc Translate(stream Msg) returns (stream Msg){}
}

message Msg
{
 string msg = 1;
}

假设将从客户端发送的消息异步发送（由于用户选择了一些 ui 元素）。

为客户端生成的 python 存根将包含一个方法Translate，该方法将接受生成器函数并返回一个迭代器。

我不清楚的是我将如何编写生成器函数来返回用户创建的消息。在等待消息时睡在线程上听起来不是最好的解决方案。

tl;dr：如何最大限度地增加可以并行发送的 http 请求数？

我正在使用aiohttp库从多个 url 获取数据。我正在测试它的性能，我观察到在这个过程中的某个地方存在瓶颈，一次运行更多的 url 无济于事。

我正在使用此代码：

import asyncio
import aiohttp

async def fetch(url, session):
    headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.3; Win64; x64; rv:64.0) Gecko/20100101 Firefox/64.0'}
    try:
        async with session.get(
            url, headers=headers, 
            ssl = False, 
            timeout = aiohttp.ClientTimeout(
                total=None, 
                sock_connect = 10, 
                sock_read = 10
            )
        ) as response:
            content = await response.read()
            return (url, 'OK', content)
    except Exception as e:
        print(e)
        return (url, 'ERROR', str(e))

async def run(url_list):
    tasks = []
    async with …

我希望我的代码使用 pythonlogging来记录异常。在我通常使用的代码中await，通常会引发异常，因此：

try: await code_that_can_raise() except Exception as e: logger.exception("Exception happended")

工作正常。

然而，当使用 loop.create_task(coro())

我不确定如何在此处捕获异常。
包装 create_task() 调用显然不起作用。在代码中记录每个异常的最佳解决方案是什么？

我需要创建一个 Observable 流，它定期发出异步协程的结果。

intervalRead是一个函数，它返回一个 Observable，并将间隔rate和异步协程函数作为参数fun，需要在定义的间隔内调用。

我的第一个方法是用interval factory方法创建一个observable，然后用map调用协程，用from_future把它包装成一个Observable，然后得到协程返回的值。

async def foo():
    await asyncio.sleep(1)
    return 42

def intervalRead(rate, fun) -> Observable:
    loop = asyncio.get_event_loop()
    return rx.interval(rate).pipe(
        map(lambda i: rx.from_future(loop.create_task(fun()))),
    )

async def main():
    obs = intervalRead(5, foo)
    obs.subscribe(
        on_next= lambda item: print(item)
    )

loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()

然而我得到的输出不是协程的结果，而是 from_future 返回的 Observable，在指定的时间间隔发出

输出： <rx.core.observable.observable.Observable object at 0x033B5650>

我怎么能得到那个 Observable 返回的实际值？我希望 42

我的第二种方法是创建一个自定义的 observable：

def intervalRead(rate, fun) -> rx.Observable:
    interval = rx.interval(rate)
    def subs(observer: Observer, scheduler = None): …