我有一个 asyncio 应用程序，其中发生了一些异常。

让我们以这个小脚本为例：

import asyncio

async def coro1():
    for a in range(1, -1, -1):
       await asyncio.sleep(1)
       print("coro1", a)
       c = 1/a  # ### !!!Will raise exception when a is 0!!!!

async def coro2():
    for n in range(6):
        print("------- coro2", n)
        await asyncio.sleep(0.5)

async def main():
    coros = [coro1(), coro2()]
    completed, pending = await asyncio.wait(coros)
    print("comp", completed)
    print("pend", pending)

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

如果我对单线程非异步程序遇到类似的问题，我就会这样做

python -m pydb scriptname

输入c并在代码中以 pydb 结束，这引发了异常，现在我可以查看和显示变量，我可以看到例如，那a是 0

我在我的 asyncio 应用程序中尝试了同样的操作，

我仅在 …

我正在尝试学习 asyncio 的用法，但遇到了障碍。

我想做什么？我正在尝试创建其中的一些，workers一旦它们被创建，他们就开始自己的task. 因此，在task3创建并启动时task1应该已经在执行其任务。我通过在单个 内使用循环来做到这一点coroutine，在每次迭代时worker创建并启动。

我面临的问题：当第一个工作人员完成其任务时，其他工作人员只是停止而不继续。

这是我的代码：

import asyncio

class Worker:
    def __init__(self, session_name):
        self.name = session_name
        self.messagelist = ['--------1', '--------2', '--------3', '--------4']

    async def job(self):
        for i, message in enumerate(self.messagelist):
            print(f"### Worker {self.name} says {message}")
            await asyncio.sleep(20)

class Testmanager:
    def __init__(self):
        self.workers_name = ['test0', 'test1', 'test2', 'test3', 'test4']

    async def create_and_start_workers(self, loop):
        for i, name in enumerate(self.workers_name):
            worker = Worker(name)
            print(f"# Created worker {worker.name}")
            loop.create_task(worker.job())
            print(f"## …

我终于开始学习 python 了asyncio，并且正在创建一个简单的客户端服务器。我可以轻松地使用open_connection和start_server并让一切都在说话。但是，当一侧意外关闭连接时，我遇到了一些意外行为。

在客户端，我可以轻松检测到服务器关闭了连接，因为任何对StreamReader.read等待的调用要么出错，要么不返回任何内容。然而，无论对方是否仍然打开套接字，StreamWriter.write/调用似乎都会成功，这让我感到困惑。await StreamWriter.drain

客户端（在协程中，通过以下方式调用run_until_complete：

_, writer = await asyncio.open_connection('127.0.0.1', 1234)
_ = input('Connected.  Type and press enter to attempt send.')

try:
    writer.write(b'A message')
    await writer.drain()
    print('Sent successfully.')

except:
    print(traceback.print_exc())
finally:
    writer.close()
    await writer.wait_closed()

服务器（同步版本，我也尝试过异步服务器）：

with socket.socket() as s:
    s.bind(('0.0.0.0', 1234))
    s.listen(1)

    con, addr = s.accept()
    with con:
        print(f'Connected to {addr[0]}:{addr[1]}')
        _ = input('Type and press Enter to close.')

如果我启动服务器，连接客户端，然后让服务器终止连接，然后告诉客户端写入，我希望根据文档和此 StackOverflow 帖子收到异常，但客户端只是打印Sent successfully. …

我正在使用 websocket 在 python 中发送和接收消息。我使用“websocket.send(msg)”以这些形式发送消息：

await ws.send(message)

和

asyncio.run(ws.send(message))

在 while 循环中，我首先检查连接是否处于活动状态，然后使用这些命令发送消息。在所有这些中，如果发送次数较少，没有问题，但是当发送次数增加时，我会收到发送消息的异常

Task <Task pending coro=<RunSocket() running at <ipython-input-1-b17eaf75a3de>:182> cb=[_run_until_complete_cb() at D:\Anaconda\InstallationFolder\lib\asyncio\base_events.py:158]> got Future <Future pending> attached to a different loop

“请注意，RunSocket 是我的函数名称之一”

然后我得到这个错误：

got Future <Future pending> attached to a different loop

我也尝试过这段代码：

asyncio.ensure_future(await ws.send(message))

但它没有发送任何消息。谁能帮我解决这个错误？任何帮助将不胜感激。

使用python3.6.8：

我正在尝试编写启动时网络设备初始配置的脚本。我的脚本在“ip_addr:port”上打开 telnet 连接。连接后，脚本会使用“\n\n”刺激其连接的网络设备（模拟管理员的两个“Enter”输入）。

connection = asyncio.open_connection(
                 ip_addr,
                 port,
             )
try:
    reader, writer = await asyncio.wait_for(connection, 5)
    print(f"successfully connected to {ip_addr}:{port}")
    writer.write(b'\n\n\n')

一些设备已经配置，我除了“login:”之外，在输入“\n”时显示在读取缓冲区中。但是，如果设备尚未配置，“login:”将不会显示在缓冲区中。因此，我认为我可以使用 wait_for 和 timeout 选项来取消此操作，并继续使用另一个reader.readuntil(...)期待另一个输出。

try:
    await asyncio.wait_for(reader.readuntil(b' login: '), 3)
    print(f"{ip_addr}:{port} alredy booted")
    break
except (asyncio.TimeoutError, OSError):
    print('Nope, moving forward')
    await reader.readuntil(b'normal setup ?(yes/no)[n]: ')

然而这会引发运行时错误。阅读文档时，我发现如果达到超时，任务就会被取消，那么为什么它不能有另一个协程readuntil()呢？

我有一个监听特定端口的异步函数。我想一次在几个端口上运行该函数，当用户想要停止在特定端口上侦听时，停止在该端口上侦听的函数。

之前我使用 asyncio 库来完成此任务，并通过创建以唯一 id 作为名称的任务来解决此问题。

asyncio.create_task(Func(), name=UNIQUE_ID)

由于 trio 使用 Nursery 来生成任务，我可以使用 Nursery.child_tasks 查看正在运行的任务，但这些任务没有办法命名它们，甚至没有办法按需取消任务

长话短说

由于 trio 没有取消特定任务的 cancel() 函数，因此如何手动取消任务。

我有一个使用 async io 对不同服务执行并行请求的应用程序。这是一个用Gunicorn运行的flask APP。

不幸的是，有些请求有点长（长达 10 秒）。到目前为止，我一直在使用 Gunicorn 的基础工作人员（同步），但由于它们的数量有限，我有时会用完它们。

所以我听说过 gevent 工作类，对于大多数请求，它允许我并行处理，但我不明白应该如何使用 asyncio 处理代码。我用这个简单的例子重现了我的问题：

我使用这个命令来启动服务器：

gunicorn test_wsgi:app --config=test_wsgi_config.py

使用 test_wsgi.py：

import asyncio

from flask import Flask

app = Flask(__name__)


async def a_long_task():
    await asyncio.sleep(5)


@app.route('/')
def hello():
    loop = asyncio.get_event_loop()

    loop.run_until_complete(
        loop.create_task(a_long_task())
    )

    return f'Hello, world'

和 test_wsgi_config.py

worker_class = "gevent"

当我使用worker_class =sync时，它工作正常，但所有请求都排队。但有了 gevent，我一直有：

RuntimeError: There is no current event loop in thread 'DummyThread-1'

如果我创建一个事件循环：

@app.route('/')
def hello():
    asyncio.set_event_loop(asyncio.new_event_loop())
    loop = asyncio.get_event_loop()

    loop.run_until_complete(
        loop.create_task(a_long_task())
    )

    return f'Hello, world'

我得到： …

要解决的问题（简化）

假设我有 26 个任务要并行运行。为了最大限度地减少服务器上的负载，我决定一次运行 10 个任务：首先并行运行 10 个任务，然后是接下来的 10 个任务，最后是剩余的 6 个任务。

我编写了一个简单的脚本来实现此行为：

import asyncio
from string import ascii_uppercase
from typing import List

TASK_NAMES = ascii_uppercase  # 26 fake tasks in total


class BatchWorker:
    """Run a list of tasks in batch."""

    BATCH_SIZE = 10

    def __init__(self, tasks: List[asyncio.Task]):
        self._tasks = list(tasks)

    @property
    def batch_of_tasks(self):
        """Yield all tasks by chunks of `BATCH_SIZE`"""
        start = 0
        while 'there are items remaining in the list':
            end = start + self.BATCH_SIZE
            chunk = self._tasks[start:end] …

当我使用mysql_poolfromaiomysql更新一条数据时，第一次和第二次是一样的。

class Test(object):\n    async def _pool(self):\n        self.pool = await aiomysql.create_\xc3\xa7(**mysql_options)\n\n    async def get_one(self, sql, param=None):\n        await self.cur.execute(sql, param)\n        result = await self.cur.fetchone()\n        return result\n\n    async def get(self):\n        self.conn = await self.pool.acquire()\n        self.cur = await self.conn.cursor(DictCursor)\n        sql = \'\'\'select policy from tb_user where id = 2;\'\'\'\n        res = await self.get_one(sql)\n        print(res)\n        await self.cur.close()\n        await self.pool.release(self.conn)\n\n    @staticmethod\n    def update():\n        import pymysql\n        coon = pymysql.connect(host=\'127.0.0.1\',\n                               port=3306,\n                               user=mysql_options[\'user\'],\n                               autocommit=True,\n                               password=mysql_options[\'password\'],\n                               database=mysql_options[\'db\'])\n        cursor = coon.cursor()\n        sql = \'\'\'update tb_user set …

我正在尝试使用 Python 中的 websockets 模块创建一个简单的 WebSocket 服务器。

我的任务是创建一个 WebSocket 服务器，向所有正在侦听它的客户端发送消息。问题是，每次客户端断开连接时，websockets 都会启动一个异常（ConnectionClosedOK或ConnectionClosedError，取决于发生的情况），我需要捕获该异常，因为我必须取消注册已断开连接的客户端，以便 WebSocket 停止在以太中发送消息（并停止WebSocket 服务器每次尝试向已关闭的客户端发送消息时都会抛出这些异常）。

这是服务器的代码：

async def foo_call():
    await asyncio.sleep(3)
    print('Hi!')


async def register(websocket):
    users.add(websocket)
    await notify_users()


async def unregister(websocket):
    users.remove(websocket)


async def start(websocket, path):
    await register(websocket)

    try:
        await foo_call()
    except websockets.ConnectionClosedError:
        await unregister(websocket)


# WebSocket clients that are listening this WebSocket server
users = set()

# Open WebSocket server
start_server = websockets.serve(start, "localhost", 8765)

# Run asyncio event loop
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

这是回溯：

Task exception was …