尝试使用 `asyncio` 模块实现 2 个“线程”

Question

我之前在 Python 中玩过线程，但决定尝试 asyncio 模块，特别是因为您可以取消正在运行的任务，这似乎是一个不错的细节。但是，由于某种原因，我无法理解它。

这是我想要实现的（抱歉，如果我使用了不正确的术语）：

• downloader 线程每 x 秒下载同一个文件，检查其哈希值是否与之前下载的内容不同，如果不同则保存。
• 在后台运行的webserver 线程，允许控制（暂停、列出、停止）downloader 线程。

我使用aiohttp 作为网络服务器。

这是我目前所拥有的：

class aiotest():

    def __init__(self):
        self._dl = None     # downloader future
        self._webapp = None # web server future
        self.init_server()

    def init_server(self):

        print('Setting up web interface')
        app = web.Application()
        app.router.add_route('GET', '/stop', self.stop)
        print('added urls')
        self._webapp = app

    @asyncio.coroutine
    def _downloader(self):
        while True:
            try:
                print('Downloading and verifying file...')
                # Dummy sleep - to be replaced by actual code
                yield from asyncio.sleep(random.randint(3,10))
                # Wait a predefined nr of seconds between downloads
                yield from asyncio.sleep(30)
            except asyncio.CancelledError:
                break

    @asyncio.coroutine
    def _supervisor(self):

        print('Starting downloader')
        self._dl = asyncio.async(self._downloader())

    def start(self):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(self._supervisor())
        loop.close()

    @asyncio.coroutine
    def stop(self):
        print('Received STOP')
        self._dl.cancel()
        return web.Response(body=b"Stopping... ")

这个类被调用：

t = aiotest()
t.start()

这当然行不通，我觉得这是一段可怕的代码。

我不清楚的地方：

• 我在stop() 方法中停止了downloader，但是我将如何停止网络服务器（例如在shutdown() 方法中）？
• downloader 是否需要新的事件循环，或者我可以使用asyncio.get_event_loop() 返回的循环吗？
• 我真的需要像supervisor 这样的东西来实现我想要实现的功能吗？这看起来很笨拙。我如何让supervisor 继续运行，而不是像现在这样在单次执行后结束？

最后一个更普遍的问题：asyncio 是否应该取代threading 模块（将来）？还是每个都有自己的应用程序？

感谢所有的指点、评论和澄清！

Answer 1

为什么当前代码不起作用：

• 您正在运行事件循环，直到 self._supervisor() 完成。 self._supervisor() 创建任务（立即发生）并立即完成。

• 您正在尝试运行事件循环直到_supervisor 完成，但是您将如何以及何时启动服务器？我认为事件循环应该一直运行到服务器停止。 _supervisor 或其他东西可以作为任务添加（到同一个事件循环）。 aiohttp 已经具有启动服务器和事件循环的功能 - web.run_app，但我们可以做到 manually。

您的问题：

1. 您的服务器将一直运行，直到您停止它。您可以启动/停止不同的 服务器工作时的协程。

2. 对于不同的协程，您只需要一个事件循环。

3. 我认为你不需要supervisor。

4. 更一般的问题：asyncio 帮助您运行不同的 在单个进程中在单个线程中并行执行功能。这就是为什么 asyncio 非常酷和快速。您的一些与线程同步的代码 可以使用 asyncio 重写，它是协程。而且：asyncio 可以 interact 带有线程和进程。 如果您仍然需要线程和进程，它会很有用：这里是 example。

有用的笔记：

• 当我们谈论非线程的异步协程时，最好使用术语 coroutine 而不是 thread
• 如果你使用Python 3.5，你可以使用async/awaitsyntax 而不是coroutine/yield from

我重写了您的代码以向您展示想法。如何查看：运行程序，查看控制台，打开http://localhost:8080/stop，查看控制台，打开http://localhost:8080/start，查看控制台，输入CTRL+C。

import asyncio
import random
from contextlib import suppress

from aiohttp import web


class aiotest():
    def __init__(self):
        self._webapp = None
        self._d_task = None
        self.init_server()

    # SERVER:
    def init_server(self):
        app = web.Application()
        app.router.add_route('GET', '/start', self.start)
        app.router.add_route('GET', '/stop', self.stop)
        app.router.add_route('GET', '/kill_server', self.kill_server)
        self._webapp = app

    def run_server(self):
        # Create server:
        loop = asyncio.get_event_loop()
        handler = self._webapp.make_handler()
        f = loop.create_server(handler, '0.0.0.0', 8080)
        srv = loop.run_until_complete(f)
        try:
            # Start downloader at server start:
            asyncio.async(self.start(None))  # I'm using controllers here and below to be short,
                                             # but it's better to split controller and start func
            # Start server:
            loop.run_forever()
        except KeyboardInterrupt:
            pass
        finally:
            # Stop downloader when server stopped:
            loop.run_until_complete(self.stop(None))
            # Cleanup resources:
            srv.close()
            loop.run_until_complete(srv.wait_closed())
            loop.run_until_complete(self._webapp.shutdown())
            loop.run_until_complete(handler.finish_connections(60.0))
            loop.run_until_complete(self._webapp.cleanup())
        loop.close()

    @asyncio.coroutine
    def kill_server(self, request):
        print('Server killing...')
        loop = asyncio.get_event_loop()
        loop.stop()
        return web.Response(body=b"Server killed")

    # DOWNLOADER
    @asyncio.coroutine
    def start(self, request):
        if self._d_task is None:
            print('Downloader starting...')
            self._d_task = asyncio.async(self._downloader())
            return web.Response(body=b"Downloader started")
        else:
            return web.Response(body=b"Downloader already started")

    @asyncio.coroutine
    def stop(self, request):
        if (self._d_task is not None) and (not self._d_task.cancelled()):
            print('Downloader stopping...')
            self._d_task.cancel()            
            # cancel() just say task it should be cancelled
            # to able task handle CancelledError await for it
            with suppress(asyncio.CancelledError):
                yield from self._d_task
            self._d_task = None
            return web.Response(body=b"Downloader stopped")
        else:
            return web.Response(body=b"Downloader already stopped or stopping")

    @asyncio.coroutine
    def _downloader(self):
        while True:
            print('Downloading and verifying file...')
            # Dummy sleep - to be replaced by actual code
            yield from asyncio.sleep(random.randint(1, 2))
            # Wait a predefined nr of seconds between downloads
            yield from asyncio.sleep(1)


if __name__ == '__main__':
    t = aiotest()
    t.run_server()