如何在Python中使用“loop.run_until_complete()”在多个异步函数之间交替?

ecbunoof  于 2022-12-01  发布在  Python
关注(0)|答案(2)|浏览(170)

我尝试在Python中交替运行2个异步函数test1()test2()withloop.run_until_complete(),如下所示:

import asyncio

async def test1():
    for _ in range(3):
        print("Test1")
        await asyncio.sleep(1)
        
async def test2():
    for _ in range(3):
        print("Test2")
        await asyncio.sleep(1)

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

loop.run_until_complete(test1()) # Here
loop.run_until_complete(test2()) # Here

但如下所示,它们并不与**loop.run_until_complete()**交替运行:

Test1
Test1
Test1
Test2
Test2
Test2

我知道如果我使用loop.run_forever()loop.create_task(),如下所示:

import asyncio

async def test1(loop):
    for _ in range(3):
        print("Test1")
        await asyncio.sleep(1)
        loop.stop() # Extra code to stop "loop.run_forever()"
        
async def test2(loop):
    for _ in range(3):
        print("Test2")
        await asyncio.sleep(1)
        loop.stop() # Extra code to stop "loop.run_forever()"

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

loop.create_task(test1(loop)) # Here
loop.create_task(test2(loop)) # Here
loop.run_forever() # Here

我可以交替运行它们,如下所示,但是loop.run_forever()永远运行,所以要停止loop.run_forever()需要额外的代码loop.stop(),这很麻烦。另外,我知道asyncio.gather()也可以交替运行它们,但是它需要await,这是我不想要的:

Test1
Test2
Test1
Test2
Test1
Test2

那么,我如何用**loop.run_until_complete()**交替运行它们呢?

zysjyyx4

zysjyyx41#

如果您不坚持使用loop.run_until_complete,则可以使用asyncio.gather功能来实现您想要的功能,如下所示:

import asyncio

async def test1():
    for _ in range(3):
        print("Test1")
        await asyncio.sleep(1)

async def test2():
    for _ in range(3):
        print("Test2")
        await asyncio.sleep(1)

async def main():
    tasks = [test1(), test2()]
    new_items = await asyncio.gather(*tasks)
    return new_items

if __name__ == '__main__':
    results = asyncio.run(main())

结果会如你所料

==> python3 stack_overflow.py 
Test1
Test2
Test1
Test2
Test1
Test2
lfapxunr

lfapxunr2#

通过使用loop.run_until_complete调用中间异步函数call_tests(),可以交替运行2个异步函数test1()test2()()如下所示.还需要使用asyncio.get_running_loop()循环。最后一个循环需要**call_tests()await中的create_task()。create_task()**交替运行它们,如下所示:

import asyncio

async def test1():
    for _ in range(3):
        print("Test1")
        await asyncio.sleep(1)
        
async def test2():
    for _ in range(3):
        print("Test2")
        await asyncio.sleep(1)

async def call_tests(): # Here
    loop = asyncio.get_running_loop() # Here
    loop.create_task(test1()) # Here
    await loop.create_task(test2()) # "await" is needed

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

loop.run_until_complete(call_tests()) # Call "call_tests()"

最后,您可以交替运行它们,如下所示:

Test1
Test2
Test1
Test2
Test1
Test2

注意,如果使用**awaitfor第一个和最后一个循环。create_task()两者都**如下所示:

# ...

async def call_tests():
    loop = asyncio.get_running_loop()
    await loop.create_task(test1()) # Here
    await loop.create_task(test2()) # Here

# ...

您不能交替运行它们,如下所示:

Test1
Test1
Test1
Test2
Test2
Test2

并且,如果使用**awaitfor第一个循环.create_task()**如下所示:

# ...

async def call_tests():
    loop = asyncio.get_running_loop()
    await loop.create_task(test1()) # Here
    loop.create_task(test2())

# ...

您不能交替运行它们,并且**最后一个loop.create_task()**在未完成的情况下退出,如下所示:

Test1
Test1
Test1
Test2

并且,如果不使用**awaitfor第一个和最后一个循环.create_task()两者都**,如下所示:

# ...

async def call_tests():
    loop = asyncio.get_running_loop()
    loop.create_task(test1()) # No "await"
    loop.create_task(test2()) # No "await"

# ...

您可以交替运行它们,但第一个和最后一个loop.create_task()在未完成的情况下退出,如下所示:

Test1
Test2

此外,如果使用**asyncio.gather(),则不需要call_tests()**,如下所示:

import asyncio

async def test1():
    for _ in range(3):
        print("Test1")
        await asyncio.sleep(1)
        
async def test2():
    for _ in range(3):
        print("Test2")
        await asyncio.sleep(1)

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

loop.run_until_complete(asyncio.gather(test1(), test2()))

然后,您可以交替运行它们,如下所示:

Test1
Test2
Test1
Test2
Test1
Test2

相关问题