Python非阻塞服务器

Question

我试图编写一个简单的非阻塞 HTTP 服务器。但我无法让所有程序（Task1，Task2 和服务器）同时运行。无论我做什么，服务器都会被阻塞。

import asyncio
from aiohttp import web


async def Task1():
    for i in range(100):
        print ('Task-1',i)
        await asyncio.sleep(1)


async def Task2():
    for i in range(100):
        print ('Task-2',i)
        await asyncio.sleep(2)

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return web.Response(text=text)



app = web.Application()
app.add_routes([web.get('/', handle), web.get('/{name}', handle)])


loop=asyncio.new_event_loop()
loop.create_task(Task1())
loop.create_task(Task2())
loop.create_task(web.run_app(app)) #with this line commented task1/2 works
loop.run_forever()

当服务器运行时，应该在终端上打印预期结果。 但是我只得到了终端输出或者服务器运行（从末尾的第二行注释掉）, 但我希望两者都有。

Task-1 0
Task-2 0
Task-1 1
Task-2 1
Task-1 2
Task-1 3
Task-2 2
Task-1 4
Task-1 5
Task-2 3
Task-1 6
---more--- 

- eSlavko

---

任务1、任务2和服务器应该“并行”运行。但是服务器却阻塞了，不允许任务1和任务2运行。 - eSlavko

你是否试图在不使用create_task的情况下调用web.run_app(app)？ - amirouche

无法正常工作。他们阻止了loop.run_forever()。 - eSlavko

我不熟悉 aiohttp，但是浏览了一下文档，有一些建议是应该使用aiohttp.web.AppRunner代替run_app。例如参考这个注释。 - larsks

Answer 1

run_app 是一个方便函数，它设置服务器并运行事件循环，直到服务器关闭。这是一个同步函数，适用于简单的示例，因此不应该传递给 create_task。唯一的原因是因为 run_app 从不返回，所以最终的 create_task 实际上并未被调用，因此 create_task 不会引发异常。

为了控制事件循环并向其中添加其他任务，可以使用 AppRunner 启动服务器。例如（未经测试）：

async def main():
    # create the application, as before
    app = aiohttp.web.Application()
    app.add_routes([
        aiohttp.web.get('/', handle),
        aiohttp.web.get('/{name}', handle)
    ])

    # add some tasks into the current event loop
    asyncio.create_task(Task1())
    asyncio.create_task(Task2())

    # set up the web server
    runner = aiohttp.web.AppRunner(app)
    await runner.setup()
    await aiohttp.web.TCPSite(runner).start()

    # wait forever, running both the web server and the tasks
    await asyncio.Event().wait()

asyncio.run(main())

- user4815162342

Answer 2

我需要做一些小的修改就可以让它正常运行...

import asyncio
from aiohttp import web


async def Task1():
    for i in range(100):
        print ('Task-1',i)
        await asyncio.sleep(1)


async def Task2():
    for i in range(100):
        print ('Task-2',i)
        await asyncio.sleep(2)



async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return web.Response(text=text)

async def HttspServer():
    # create the application, as before
    app = web.Application()
    app.add_routes([
        web.get('/', handle),
        web.get('/{name}', handle)
    ])

    # set up the web server
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner)
    await site.start()

    # wait forever, running both the web server and the tasks
    await asyncio.Event().wait()


loop=asyncio.new_event_loop()
loop.create_task(Task1())
loop.create_task(Task2())
loop.create_task(HttspServer())
loop.run_forever()

- eSlavko

---

请注意，asyncio.run() 是启动 asyncio 应用程序的推荐方式。如果您不使用 asyncio.run，则不需要在 HtspServer 结尾处使用 await asyncio.Event().wait() - 这只是我版本中为了防止 main() 返回并导致程序退出而添加的。 - user4815162342

我不能按照你的方式添加asyncio.create_task(Task1())。但是，我的方式是否存在一些我不知道的真正缺点呢？因为它至少似乎没有出现问题。 - eSlavko

抱歉，您可以将 asyncio.create_task(x) 替换为 loop.create_task(x)，其中 loop 是通过 loop = asyncio.get_event_loop() 获得的。在当前的 Python 中，使用 run_forever 方法没有实际的缺点，但是您可能会想要稍后迁移到 asyncio.run()，那时您的方法将不起作用，因为 asyncio.run() 会创建一个新的事件循环，而不像 run_until_complete 和 run_forever。 - user4815162342

我在使用asyncio.get_event_loop()时遇到了麻烦，因为我的程序中所有的东西都在类内部。据我所知，如果不从主程序执行，asyncio.get_event_loop()将无法正常工作。 - eSlavko

您始终可以在协程(async def)中使用asyncio.get_event_loop()。这就是我所指的。 - user4815162342

Answer 3

这应该是最终解决方案了吗？

#!/usr/bin/python
# -*- coding: utf8 -*-
import asyncio
from aiohttp import web


async def Task1():
    for i in range(100):
        print ('Task-1',i)
        await asyncio.sleep(1)

async def Task2():
    for i in range(100):
        print ('Task-2',i)
        await asyncio.sleep(2)

async def handle(request):
    data = {'some': 'data'}
    return web.json_response(data)

async def main():
    # create the application, as before
    app = web.Application()
    app.add_routes([
        web.get('/', handle),
        web.get('/{name}', handle)
    ])

    # add some tasks into the current event loop
    loop = asyncio.get_event_loop()
    loop.create_task(Task1())
    loop.create_task(Task2())

    # set up the web server
    runner = web.AppRunner(app)
    await runner.setup()
    await web.TCPSite(runner).start()

    # wait forever, running both the web server and the tasks
    await asyncio.Event().wait()

asyncio.get_event_loop().run_until_complete(main())

- eSlavko

---

是的，看起来很棒，并且完全兼容将来在顶层使用 asyncio.run()。请注意，您可以编辑先前的答案，您不需要发布新的答案来更新或更正它。 - user4815162342

我本来想那么做，但后来我将其作为新程序添加以保持从何到何的工作流程。虽然可以更易读，但缺点是需要读更多内容。 - eSlavko

一旦Task1和Task2完成，您如何停止Web服务器？ - toryan