标签: python-asyncio

我有几个函数，它们的执行并不相互依赖。我想做的是同时执行它们而不是顺序执行（同步）。我也添加了事件循环，但我无法确定它是否正常工作。

这是实现：

文件一：

import file2

def funcA():
    a, b = 1, 2
    file2.main(a, b)

文件2：

def main(a, b):
    asyncio.get_event_loop().run_until_complete(_main(a, b))

async def _main(a, b):
     out1 = await funcA(a, b)
     out2 = await funcB(a, b)
     out3 = await funcC(a, b)

async def funcA(a, b):
    result = 1 # some processing done here
    return result

async def funcB(a, b):
    result = 1 # some processing done here
    return result

async def funcC(a, b):
    result = 1 # some processing done here
    return …

有什么区别？

asyncio.run(func())

和

loop = asyncio.get_event_loop()
loop.run_until_complete(func())

据我了解，区别在于，如果我在已有的程序中使用第一个示例event_loop，则会发生冲突。

为什么 asyncio.gather 不能与生成器表达式一起使用？

import asyncio

async def func():
    await asyncio.sleep(2)

# Works
async def call3():
    x = (func() for x in range(3))
    await asyncio.gather(*x)

# Doesn't work
async def call3():
    await asyncio.gather(func() for x in range(3))

# Works
async def call3():
    await asyncio.gather(*[func() for x in range(3)])

asyncio.run(call3())

第二个变体给出：

[...]
  File "test.py", line 13, in <genexpr>
    await asyncio.gather(func() for x in range(3))
RuntimeError: Task got bad yield: <coroutine object func at 0x10421dc20>

这是预期的行为吗？

我正在编写一个示例程序，它从数据源（csv 或 rdbms）中读取数据块，进行一些转换并通过套接字将其发送到服务器。

但因为 csv 非常大，出于测试目的，我想在几个块之后中断读取。不幸的是，出了问题，我不知道是什么以及如何解决它。也许我必须取消一些活动，但现在确定在哪里以及如何取消。我收到以下错误：

Task was destroyed but it is pending!
task: <Task pending coro=<<async_generator_athrow without __name__>()>>

示例代码是：

import asyncio
import json

async def readChunks():
  # this is basically a dummy alternative for reading csv in chunks
  df = [{"chunk_" + str(x) : [r for r in range(10)]} for x in range(10)]
  for chunk in df:
    await asyncio.sleep(0.001)
    yield chunk

async def send(row):
    j = json.dumps(row)
    print(f"to be sent: {j}")
    await asyncio.sleep(0.001)


async def main():
    i = 0 …

我是Tor 的 python 库Stem的作者。最近我们迁移到 asyncio，但希望同步调用者仍然可以使用。

我们编写了一个 mixin，它可以透明地使任何类都可以由异步和同步调用者使用......

import asyncio
import stem.util.asyncio

class Example(stem.util.asyncio.Synchronous):
  async def hello(self):
    return 'hello'

def sync_demo():
  instance = Example()
  print('%s from a synchronous context' % instance.hello())
  instance.stop()

async def async_demo():
  instance = Example()
  print('%s from an asynchronous context' % await instance.hello())
  instance.stop()

sync_demo()
asyncio.run(async_demo())

有更好的方法来解决这个问题吗？

我们更广泛的生态系统似乎正在将网络库分成同步/异步变体，其中请求和AIOHTTP是最突出的。是否有任何库可以在不重复其 API 的情况下对这两种调用者提供良好的支持？

谢谢！

例子

这是我的代码：

from typing import  List
from fastapi import FastAPI, File, UploadFile
import asyncio
import concurrent.futures

app = FastAPI()
@app.post("/send_images")
async def update_item(
    files: List[UploadFile] = File(...),
):
    return {"res": len(files)}

我向该服务器发送请求（这些特定网址仅作为示例）：

import requests
import os 
import json
import numpy as np
import time
import io
import httpx
import asyncio
from datetime import datetime
import pandas as pd

from random import shuffle
import pandas as pd
import urllib
import io
from PIL import Image
from matplotlib import pyplot as plt …

据我了解，目的是同时asyncio.gather运行其参数，并且当协程执行等待表达式时，它为事件循环提供了安排其他任务的机会。考虑到这一点，我惊讶地发现以下代码片段忽略了.asyncio.gather

import asyncio                                                             
  
async def aprint(s):
    print(s)

async def forever(s):
    while True:
        await aprint(s)

async def main():
    await asyncio.gather(forever('a'), forever('b'))

asyncio.run(main())

async def forever(s):
    while True:
        await aprint(s)
        await asyncio.sleep(1.)

async def forever(s):
    while True:
        await aprint(s)
        await asyncio.sleep(0.)