标签: python-asyncio

我有一个关于 pythonasyncio模块中的事件循环如何管理未完成任务的问题。考虑以下代码：

import asyncio

@asyncio.coroutine
def a():
   for i in range(0, 3):
      print('a.' + str(i))
      yield


@asyncio.coroutine
def b():
   for i in range(0, 3):
      print('b.' + str(i))
      yield


@asyncio.coroutine
def c():
   for i in range(0, 3):
      print('c.' + str(i))
      yield


tasks = [
   asyncio.Task(a()),
   asyncio.Task(b()),
   asyncio.Task(c()),
]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait([t1, t2, t3]))

运行这将打印：

a.0
b.0
c.0
a.1
b.1
c.1
a.2
b.2
c.2

请注意，它总是打印出 'a' 然后是 'b' 然后是 'c'。我猜无论每个协程经过多少次迭代，它都将始终按该顺序打印。所以你永远不会看到类似的东西

b.100
c.100
a.100

来自 node.js 背景，这告诉我这里的事件循环在内部维护一个队列，它用来决定接下来运行哪个任务。它最初放在a()队列的前面，然后是b() …

我有一个像傻瓜一样的代码：

def render():
    loop = asyncio.get_event_loop()

    async def test():
        await asyncio.sleep(2)
        print("hi")
        return 200

    if loop.is_running():
        result = asyncio.ensure_future(test())
    else:
        result = loop.run_until_complete(test())

当loop没有运行时很容易，只需使用loop.run_until_complete它并返回 coro 结果，但如果循环已经在运行（我的阻塞代码在已经运行循环的应用程序中运行）我不能使用，loop.run_until_complete因为它会引发异常；当我调用asyncio.ensure_future任务被安排并运行时，但我想在那里等待结果，有人知道怎么做吗？文档不是很清楚如何做到这一点。

我尝试在 coro 内部传递一个concurrent.futures.Future调用set_result，然后调用Future.result()我的阻塞代码，但它不起作用，它阻塞在那里并且不让其他任何东西运行。任何帮助，将不胜感激。

我想问一下asyncio.Condition。我不熟悉这个概念，但我从学生时代就知道并理解锁、信号量和队列。

我找不到很好的解释或典型的用例，只是这个例子。我看了一下源码。核心功能是通过期货的先进先出实现的。每个等待的协程都会添加一个新的未来并等待它。另一个协程可能会调用notify()它从 FIFO 设置一个或多个期货的结果，并唤醒相同数量的等待协程。到目前为止真的很简单。

但是，实现和使用比这更复杂。等待协程必须首先获取与条件关联的锁，以便能够等待（并wait()在等待时释放它）。此外，通知程序必须获得一个锁才能通知（）。这导致with在每个操作之前声明：

async with condition:
    # condition operation (wait or notify)

否则会发生RuntimeError。

我不明白拥有这个锁的意义。我们需要用锁保护哪些资源？在 asyncio 中，事件循环中总是只有一个协程在执行，没有线程中已知的“临界区”。

这个锁真的需要（为什么？）还是只是为了与线程代码兼容？

我的第一个想法是为了兼容性，但在这种情况下，为什么他们在保留使用的同时不删除锁？即制作

async with condition:

基本上是可选的无操作。

拥有一个异步生成器，我希望能够异步迭代它。但是，我遗漏了一些东西或搞砸了一些东西或两者兼而有之，因为我最终得到了一个常规的同步 for 循环：

import asyncio


async def time_consuming(t):
    print(f"Going to sleep for {t} seconds")
    await asyncio.sleep(t)
    print(f"Slept {t} seconds")
    return t


async def generator():
    for i in range(4, 0, -1):
        yield await time_consuming(i)


async def consumer():
    async for t in generator():
        print(f"Doing something with {t}")


if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    loop.run_until_complete(consumer())
    loop.close()

这将需要大约 12 秒来运行并返回：

Going to sleep for 4 seconds
Slept 4 seconds
Doing something with 4
Going to sleep for 3 seconds
Slept 3 seconds
Doing …

我正在使用事件asyncio forever()循环。现在我想在进程或信号或文件更改后重新启动循环（停止循环并重新创建一个新循环），但我有一些问题要做：

下面是三个简化的代码片段，其中演示了一些协程工作器和一个协程循环重启器：

#第一次尝试：

import asyncio

async def coro_worker(proc):
    print(f'Worker: {proc} started.')
    while True:
        print(f'Worker: {proc} process.')
        await asyncio.sleep(proc)

async def reset_loop(loop):
    # Some process
    for i in range(5):  # Like a process.
        print(f'{i} counting for reset the eventloop.')
        await asyncio.sleep(1)

    main(loop)  # Expected close the current loop and start a new loop!

def main(previous_loop=None):
    offset = 0
    if previous_loop is not None:  # Trying for close the last loop if exist.
        offset = 1  # An offset …

我正在阅读 Python 文档和PyMotW书籍，试图学习 Async/ Await、Futures 和 Tasks。

协程和任务文档：

通常不需要在应用程序级代码创建 Future 对象。

从未来的文档中，它说明了以下内容：

loop.create_future()

创建一个附加到事件循环的 asyncio.Future 对象。

这是在 asyncio 中创建 Futures 的首选方式。这让第三方事件循环提供 Future 对象的替代实现（具有更好的性能或检测）。

然而，在PyMotW 关于 Future 的章节中，作者创建了一个future这样的对象：

all_done = asyncio.Future()

我猜是因为这本书比 Python 的当前版本稍有落后。为了纠正这个问题，我做了以下事情：

future_Obj = event_loop.create_future()

所以作者的完整代码变成了：

import asyncio


def mark_done(future, result):
    print('setting future result to {!r}'.format(result))
    future.set_result(result)


event_loop = asyncio.get_event_loop()

try:

    future_Obj = event_loop.create_future()
    print('scheduling mark_done')
    event_loop.call_soon(mark_done, future_Obj, 'the result')

    print('entering event loop')
    result = event_loop.run_until_complete(future_Obj)
    print('returned result: {!r}'.format(result)) …

我正在尝试确定如何通过 asyncio 代码正确使用 cProfile 和 pstats。我正在通过运行来分析我的脚本cProfile.run('loop.run_until_complete(main())', 'profile.stats')。

使用pstatsto sort by 后SortKeys.TIME，我得到以下输出：

In [9]: sorted_stats.print_stats()
Fri Feb 15 14:40:18 2019    profile.txt

         67411199 function calls (67230882 primitive calls) in 305.899 seconds

   Ordered by: internal time

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
     3617  273.162    0.076  273.162    0.076 {method 'poll' of 'select.epoll' objects}
14094092/14093931    1.527    0.000    1.904    0.000 {built-in method builtins.isinstance}
    44901    1.166    0.000    1.166    0.000 {built-in method posix.stat}
   784052    1.028    0.000    1.479    0.000 /usr/lib/python3/dist-packages/soupsieve/util.py:41(lower)
    32070    0.887    0.000 …

我正在使用 discord.py 库在 Python 3.6 中开发 Discord 机器人，并且在尝试从我创建的线程向特定通道发送消息时遇到问题。

基本上，我有一些线程监视网站，并且我想调用一个函数（我正在传递给线程），当我注意到网站上的更改时，该函数将消息发送到一个特定的频道。

我最初尝试在client.send_message()没有 async/await 的情况下直接调用但它根本不起作用，所以我编写了发送消息的 async/await 函数（测试它并且它有效）但是我再次从线程调用它时遇到问题，所以我结束了将机器人客户端传递给我的线程并调用self.bot_client.loop.create_task(self.sendmsgfunction(msg)). 这是可行的，但与机器人使用常规@bot.event功能回复消息所需的时间相比，它非常慢（发送消息大约需要 15 秒，而且这可能不是正确的发送方式）。

我已经尝试使用 asyncio 创建一个事件循环，然后调用该函数，但最终还是出现了错误。

有任何想法吗？

在我的代码中，我有一个带有属性的类，偶尔需要运行异步代码。有时我需要从异步函数访问属性，有时从同步函数访问属性 - 这就是为什么我不希望我的属性是异步的。此外，我的印象是异步属性通常是一种代码异味。如果我错了纠正我。

我在从同步属性执行异步方法并阻止进一步执行直到异步方法完成时遇到问题。

这是一个示例代码：

import asyncio


async def main():
    print('entering main')
    synchronous_property()
    print('exiting main')


def synchronous_property():
    print('entering synchronous_property')
    loop = asyncio.get_event_loop()
    try:
        # this will raise an exception, so I catch it and ignore
        loop.run_until_complete(asynchronous())
    except RuntimeError:
        pass
    print('exiting synchronous_property')


async def asynchronous():
    print('entering asynchronous')
    print('exiting asynchronous')


asyncio.run(main())

它的输出：

entering main
entering synchronous_property
exiting synchronous_property
exiting main
entering asynchronous
exiting asynchronous

首先，RuntimeError捕获似乎是错误的，但如果我不这样做，我会得到RuntimeError: This event loop is already running例外。

其次，asynchronous()函数最后执行，同步完成后。我想通过异步方法对数据集进行一些处理，所以我需要等待它完成。如果我await asyncio.sleep(0)在调用之后添加synchronous_property() …

在阅读了许多文档AsyncIO和文章后，我仍然找不到答案：异步运行函数（不使用线程）并确保调用此异步函数的函数继续执行。

伪代码：

async def functionAsync(p):
    #...
    #perform intensive calculations
    #...
    print ("Async loop done")

def functionNormal():
    p = ""
    functionAsync(p)
    return ("Main loop ended")

 print ("Start Code")
 print functionNormal()

预期输出：

Start code
Main loop ended
Async loop done

搜索使用的示例loop.run_until_complete，但不会返回 的打印值，functionNormal()因为它本质上是阻塞的。