标签: python-asyncio

我需要做很多工作，但幸运的是，很容易解耦到不同的任务中进行异步执行。其中一些是相互依赖的，我很清楚如何在任务中await与其他人一起获得结果。但是，我不知道如何让多个不同的任务等待同一个协程，并且都得到结果。据我所知，文档也没有提及这种情况。考虑以下最小示例：

from asyncio import create_task, gather

async def TaskA():
    ...  # This is clear
    return result

async def TaskB(task_a):
    task_a_result = await task_a
    ...  # So is this
    return result

async def TaskC(task_a):
    task_a_result = await task_a
    ...  # But can I even do this? 
    return result

async def main():
    task_a = create_task(TaskA())
    task_b = create_task(TaskB(task_a))
    task_c = create_task(TaskC(task_a))
    gather(task_b, task_c)  # Can I include task_a here to signal the intent of "wait for all tasks"? …

我想知道，如果 python GIL 只允许一次运行一个线程/进程，为什么我应该使用 asyncio，我知道线程之间的切换很昂贵，但是就是这样？这是Python中asyncio的唯一优点吗？

在 .Net C# 中，有一个函数Task.WhenAll可以获取任务列表来等待它们。我应该在 python 中使用什么？我正在尝试对此做同样的事情：

tasks = ...  # list of coroutines
    
for task in tasks:
    await task

Django 从 3.1 版本开始支持异步视图，因此它非常适合对外部 HTTP API 等非阻塞调用（例如使用aiohttp）。

我经常看到以下代码示例，我认为它在概念上是错误的（尽管它工作得很好）：

import aiohttp\nfrom django.http import HttpRequest, HttpResponse\n\nasync def view_bad_example1(request: HttpRequest):\n    async with aiohttp.ClientSession() as session:\n        async with session.get("https://example.com/") as example_response:\n            response_text = await example_response.text()\n            return HttpResponse(response_text[:42], content_type="text/plain")\n

此代码ClientSession为每个传入请求创建一个，效率低下。aiohttp然后不能使用例如连接池。

\n

不要\xe2\x80\x99t为每个请求创建一个会话。您很可能需要每个应用程序有一个会话来共同执行所有请求。

\n

来源：https ://docs.aiohttp.org/en/stable/client_quickstart.html#make-a-request

\n

这同样适用于 httpx：

\n

另一方面，客户端实例使用 HTTP 连接池。这意味着当您向同一主机发出多个请求时，客户端将重用底层 TCP 连接，而不是为每个请求重新创建一个连接。

\n

来源：https://www.python-httpx.org/advanced/#why-use-a-client

\n

有什么方法可以aiohttp.ClientSession在 Django 中全局实例化，以便可以在多个请求之间共享该实例吗？不要忘记必须ClientSession在正在运行的事件循环中创建（为什么在事件循环之外创建 ClientSession …

我正在使用 asyncio.gather，并在出现异常时尝试重试。但它无法重试，并出现如下错误。我应该如何修复它以重试？

11 -- RETRY Server disconnected without sending a response.
11 -- RETRY cannot reuse already awaited coroutine
11 -- RETRY cannot reuse already awaited coroutine

我的代码：

        for i in range((len(tasks) // TASK_CHUNK_SIZE) + 1):
            while True:
                try:
                    res = await asyncio.gather(*tasks[TASK_CHUNK_SIZE * i:TASK_CHUNK_SIZE * (i + 1)])
                    for rows in res:
                        f.writelines([f'{row}\n' for row in rows])
                    await asyncio.sleep(1)
                    break
                except Exception as e:
                    print(f'{(i + 1) * TASK_CHUNK_SIZE} -- RETRY {e}')

                    await asyncio.sleep(10)
                    continue

状态的 Python 文档asyncio.create_task：

\n

重要提示：保存对此函数结果的引用，以避免任务在执行过程中消失。事件循环仅保留对任务的弱引用。未在其他地方引用的任务可能会随时被垃圾回收，甚至在它完成之前也是如此。对于可靠的 \xe2\x80\x9cfire-and-forget\xe2\x80\x9d 后台任务，请将它们收集在一个集合中：

\n

asyncio在创建的任务完成之前使用弱引用而不是强引用有什么好处？正如上面的警告所表明的那样，保留强引用肯定至少有一个好处——这意味着弱引用可能会带来抵消性的好处。请注意，对于强引用，asyncio完成后可以完全删除引用或切换到弱引用，具体取决于异步逻辑的要求。

文档警告中隐含的用例是：我们不想等待任务（特别是不使用任务的返回值），但我们确实希望给任务一个最终运行的机会。

请注意，asyncio.TaskGroup上下文确实保留了对其创建的任务的强引用。但是，它不适合上述用例，因为退出时，上下文会阻塞（它等待所有任务完成）。（此外，即使任务完成后，它也会保留对任务的强引用，因此如果上下文的生存时间比任务长得多，则会导致内存泄漏。）

我想了解如何在简单的场景中使用 telnetlib3。

长期存在的 telnetlib（不是 3）在https://docs.python.org/3/library/telnetlib.html有一个简单的示例，其中 python 程序连接到 telnet 服务器，然后查找提示并提供响应。人们可以很容易地看到如何将此示例扩展到不同的提示、添加超时以及添加更多提示响应步骤。

import getpass
import telnetlib

HOST = "localhost"
user = input("Enter your remote account: ")
password = getpass.getpass()

tn = telnetlib.Telnet(HOST)

tn.read_until(b"login: ")
tn.write(user.encode('ascii') + b"\n")
if password:
    tn.read_until(b"Password: ")
    tn.write(password.encode('ascii') + b"\n")

tn.write(b"ls\n")
tn.write(b"exit\n")

print(tn.read_all().decode('ascii'))

但是，telnetlib（不是 3）已被弃用。

替代品 telnetlib3 ( https://telnetlib3.readthedocs.io/en/latest/intro.html#quick-example ) 提供了一个基于 asyncio 的示例，并且异步“shell”函数（与服务器交互）阻塞等待提示（异步的基本原理）并始终用“y”响应服务器。

import asyncio, telnetlib3

async def shell(reader, writer):
    while True:
        # read stream until '?' mark is found
        outp = await reader.read(1024)
        if not outp: …

我有示例echo服务器

import asyncio

class EchoServer(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        self.transport.write(data)
        # Client piece goes here

loop = asyncio.get_event_loop()
coro = loop.create_server(EchoServer, '127.0.0.1', 8888)
server = loop.run_until_complete(coro)
print('serving on {}'.format(server.sockets[0].getsockname()))

try:
    loop.run_forever()
except KeyboardInterrupt:
    print("exit")
finally:
    server.close()
    loop.close()

我正在尝试做的是添加一个客户端部分,我评论说它将连接到一台新服务器并从该方式发送数据.有echo客户端,但我需要一个如下所示的进程:

    +-----------+    +-----------+    +--------------+
    | My Server |    | My Client |    | Other Server |
    +-----------+    +-----------+    +--------------+
          |                |                 |
 ===>Get some data         |                 |
          |                | …

我是Python 3.5 asyncio的新手.

在下面的代码中,asyncio.wait()不等待stop_future完成:

import asyncio
import datetime
from concurrent.futures import FIRST_COMPLETED


def stop():  # callback after 12 seconds

    print('stop', datetime.datetime.now())
    stop_future.set_result('Done!')


async def display_dt():

    while not stop_future.done():
        print('dt-1', datetime.datetime.now())
        # sleep 5 seconds or stop_future done
        asyncio.wait([await asyncio.sleep(5), stop_future],  return_when=FIRST_COMPLETED)
        print('dt-2', datetime.datetime.now())

    task = asyncio.Task.current_task()
    task.cancel()
    print(stop_future.result())


loop = asyncio.get_event_loop()
stop_future = asyncio.Future()
loop.call_later(12, stop)
loop.run_until_complete(display_dt())
loop.close() 

结果:

dt-1 2015-11-08 00:49:37.324582
dt-2 2015-11-08 00:49:42.325503
dt-1 2015-11-08 00:49:42.325503
dt-2 2015-11-08 00:49:47.326423
dt-1 2015-11-08 00:49:47.326423
stop 2015-11-08 00:49:49.327192  # async.wait stop_future …

由于某种原因,该程序打印以下警告:

Task exception was never retrieved
future: <Task finished coro=<coro() done, defined at /usr/lib64/python3.4/asyncio/coroutines.py:139> exception=SystemExit(2,)>

即使异常被清楚地检索和传播,如caught SystemExit!打印到终端,并且进程状态代码变为2.

Python 2和trollius也是如此.

我错过了什么吗？

#!/usr/bin/env python3

import asyncio

@asyncio.coroutine
def comain():
    raise SystemExit(2)

def main():
    loop = asyncio.get_event_loop()
    task = loop.create_task(comain())
    try:
        loop.run_until_complete(task)
    except SystemExit:
        print("caught SystemExit!")
        raise
    finally:
        loop.close()

if __name__ == "__main__":
    main()