我编写了一个测试程序,使用create_task()进行测试,需要等到创建的任务完成.

我try 使用loop.run_until_complete()等待任务完成,但它导致了一个回溯错误.

/Users/jason/.virtualenvs/xxx/bin/python3.5 /Users/jason/asyncio/examples/hello_coroutine.py
    Traceback (most recent call last):
    Test
      File "/Users/jason/asyncio/examples/hello_coroutine.py", line 42, in <module>
    Hello World, is a task
        loop.run_until_complete(test.greet_every_two_seconds())
      File "/Users/jason/asyncio/asyncio/base_events.py", line 373, in run_until_complete
        return future.result()
      File "/Users/jason/asyncio/asyncio/futures.py", line 274, in result
        raise self._exception
      File "/Users/jason/asyncio/asyncio/tasks.py", line 240, in _step
        result = coro.send(None)
      File "/Users/jason/asyncio/examples/hello_coroutine.py", line 33, in greet_every_two_seconds
        self.a()
      File "/Users/jason/asyncio/examples/hello_coroutine.py", line 26, in a
        t = loop.run_until_complete(self.greet_every_one_seconds(self.db_presell))
      File "/Users/jason/asyncio/asyncio/base_events.py", line 361, in run_until_complete
        self.run_forever()
      File "/Users/jason/asyncio/asyncio/base_events.py", line 326, in run_forever
        raise RuntimeError('Event loop is running.')
    RuntimeError: Event loop is running.

测试代码如下.函数a()不能是协同程序,

我怎样才能等到任务完成?

import asyncio

class Test(object):

    def __init__(self):
        print(self.__class__.__name__)
        pass
    @asyncio.coroutine
    def greet_every_one_seconds(self, value):
            print('Hello World, one second.')
            fut = asyncio.sleep(1,result=value)
            a = yield from fut
            print(a)

    def a(self):

        loop = asyncio.get_event_loop()

        task=loop.run_until_complete(self.greet_every_one_seconds(4))
        #How can i wait for the task until complete?

    @asyncio.coroutine
    def greet_every_two_seconds(self):
        while True:
            self.a()
            print('Hello World, two seconds.')
            yield from asyncio.sleep(2)


if __name__ == '__main__':
    test = Test()
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(test.greet_every_two_seconds())
    finally:
        loop.close()

推荐答案

我怎样才能等到任务完成?

在普通函数中为loop.run_until_complete(task).或者在一次合作中是await task.

下面是一个完整的代码示例,演示了这两种情况:

#!/usr/bin/env python3
import asyncio

async def some_coroutine(loop):
    task = loop.create_task(asyncio.sleep(1))  # just some task
    await task # wait for it (inside a coroutine)

loop = asyncio.get_event_loop()
task = loop.create_task(asyncio.sleep(1)) # just some task
loop.run_until_complete(task) # wait for it (outside of a coroutine)
loop.run_until_complete(some_coroutine(loop))

Python-3.x相关问答推荐

使用魔方无法从图像中识别单个字符

将Trio与基于线程的事件侦听器混合使用

Pandas 数据帧断言等同于NaN

按小时和日期对Pandas 数据帧进行分组

PythonPandas 创建一个列并添加到DataFrame

在Python代码中包含NAN值时,以两个矩阵计算RMSE

Python中根据分组/ID对两个数据框进行映射,以更接近值的升序排列

在Python中基于组/ID将两个数据帧进行映射,找出较接近的值

如何计算Pandas 列中每列唯一项目的出现次数?

从日志(log)文件中查找延迟最低的用户

在不使用字符串方法的情况下查找字符串最后一个单词的长度 - Python

运行 PyCharm 测试时如何解决django.core.exceptions.ImproperlyConfigured:找不到 GDAL 库?

Python - For 循环数百万行

为什么Pandas会在 NaN 上合并?

创建日志(log)文件

如何将 SimpleGUI 与 Python 2.7 和 3.0 shell 集成

在 Alembic 迁移期间更新列内容

如何替换 Python pathlib.Path 中的子字符串?

用于 unicode 大写单词的 Python 正则表达式

带有自定义标头的 urllib.urlretrieve