标签: python-asyncio

我正在使用 asyncio 进行一些 TCP 通信。我有一个无限循环的Receive()函数。read()这作为后台任务运行，使用asyncio.create_task(Receive()).

现在，如果连接被对等方关闭，则会引发异常（或者可能是任何其他异常），我在函数中捕获该异常Receive()。但是，我想重新引发该异常，以便外部代码可以决定要做什么（例如重新连接）。

由于异常是在任务中引发的，我不知道如何检索它。

我试图创建一个例子来说明我的意思：

import asyncio

async def divide(x):
    try:
        return 1/x
    except Exception as e:
        print("Divide inner exception: ", e)
        raise   # Re-raise so main() can handle it

async def someFn():
    asyncio.create_task(divide(0))  # Exception is never retrieved
    # await divide(0) # This will raise two exceptions - the original in divide() and in main()

async def main():
    try:
        await someFn()
        # Do other things while someFn() runs …

我们知道，通过request.state我们可以将一些自定义数据从中间件的处理前过程传递给处理程序，从而影响其行为，我目前想知道处理程序如何影响中间件之后的逻辑。

我的具体业务场景是，我有一个路由地址（举/api个例子），它的作用是计算一个动态结果，会消耗比较长的时间，返回比较大的json响应。从操作上来说，提高效率的一个有效方法是使用一些缓冲区（如redis）来缓存其结果，这样每次缓存命中时都可以节省计算时间。

由于我的内存有限，我想将 gzip 压缩的字节存储在缓冲区中而不是原始 json 流中，这将大大增加我可以处理的缓存量。numbers具体来说，由于响应内容较多，未经gzip处理的响应内容通常在20MB左右，而压缩后只有1MB左右。这意味着，使用 1GB 内存，在不压缩的情况下我只能缓存 50 个不同的响应，而在压缩的情况下我可以缓存 1000 个，这是一个难以忽视的显着差异。

由于这些需求，我想实现一个功能齐全的 gzip 中间件，但存在几个技术困惑。第一个是我想控制中间件是否压缩，显然，如果响应没有命中缓存而是动态生成的，那么它应该被压缩，但相反，它不应该再次压缩，因为它已经已经被压缩过一次了。第二个问题是，即使我可以控制中间件不压缩，当不需要运行压缩逻辑时，如何将其结果替换为已经压缩的字节？

由于我还不知道如何实现，请原谅我只能提供一些伪代码来说明我的想法。

下面的代码描述了一个相对复杂的不包含中间件的响应地址：

from fastapi import FastAPI, Request
import uvicorn

app = FastAPI()

@app.post("/api")
async def root(request :Request, some_args: int):

    # Using the Fibonacci series to simulate a time-consuming 
    # computational operation
    def fib_recur(n):
    if n <= 1: return n
    return fib_recur(n-1) + fib_recur(n-2)
    
    # In addition to the time-consuming calculation, the size of 
    # the returned content is …

我想使用 python 并行运行许多 HTTP 请求。我尝试使用 asyncio 这个名为 aiohttp 的模块。

import aiohttp
import asyncio

async def main():
    async with aiohttp.ClientSession() as session:
        for i in range(10):
            async with session.get('https://httpbin.org/get') as response:
                html = await response.text()
                print('done' + str(i))

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

我期望它并行执行所有请求，但它们是一一执行的。虽然，我后来使用线程解决了这个问题，但我想知道这有什么问题吗？

最近读了 asyncio 的代码，对loop.call_soon和loop.call_soon_threadsafe感到困惑。我发现的唯一区别是call_soon_threadsafe中有一个_write_to_self。众所周知，事件循环在特定线程中运行。事件循环安排的所有任务都按顺序进行。我们可以假设任务在被一一调度时是线程安全的。那么call_soon中怎么会出现线程不安全的情况呢？

这是我尝试使用高级 asyncio 子进程 API过早终止子进程的程序：

import asyncio

async def test():
    process = await asyncio.create_subprocess_shell(
        "sleep 2 && echo done",
        stdout=asyncio.subprocess.PIPE,
    )
    await asyncio.sleep(1)
    process.kill()
    await process.wait()
    # process._transport.close()

asyncio.run(test())

退出时，它将以下内容写入stderr：

$ python3.9 test.py

Exception ignored in: <function BaseSubprocessTransport.__del__ at 0x1065f0dc0>
Traceback (most recent call last):
  File "/usr/local/Cellar/python@3.9/3.9.2_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_subprocess.py", line 126, in __del__
    self.close()
  File "/usr/local/Cellar/python@3.9/3.9.2_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_subprocess.py", line 104, in close
    proto.pipe.close()
  File "/usr/local/Cellar/python@3.9/3.9.2_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/unix_events.py", line 536, in close
    self._close(None)
  File "/usr/local/Cellar/python@3.9/3.9.2_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/unix_events.py", line 560, in _close
    self._loop.call_soon(self._call_connection_lost, exc)
  File "/usr/local/Cellar/python@3.9/3.9.2_1/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 746, …

设置：

所以我有一个由客户端对象组成的客户端数组。还有两个异步函数，一个用于处理来自 WebSocket 的传入消息，另一个用于使用给定的客户端对象执行一些 API 调用。

clients = {}

async def function_a(ws_message): # this is being called each time when a new message arrives
    if ws_message == 'signal_message':
        for client in clients:
             await function_b(client)

async def function_b(client):
    # get client parameters
    # make an API request with those parameters