如何使用事件循环和执行程序关闭进程

Question

考虑以下程序。

import asyncio
import multiprocessing
from multiprocessing import Queue
from concurrent.futures.thread import ThreadPoolExecutor
import sys


def main():
    executor = ThreadPoolExecutor()
    loop = asyncio.get_event_loop()
    # comment the following line and the shutdown will work smoothly
    asyncio.ensure_future(print_some(executor))

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print("shutting down")
        executor.shutdown()
        loop.stop()
        loop.close()
        sys.exit()


async def print_some(executor):
    print("Waiting...Hit CTRL+C to abort")
    queue = Queue()
    loop = asyncio.get_event_loop()
    some = await loop.run_in_executor(executor, queue.get)
    print(some)


if __name__ == '__main__':
    main()

我只想在按“ CTRL + C”时正常关机。但是，执行程序线程似乎阻止了该操作（即使我确实致电shutdown）

Answer 1

您需要发送毒药以使工作人员停止在queue.get呼叫上收听。ThreadPoolExecutor如果池中的工作线程有活跃的工作，它们将阻止Python退出。在源代码中有一条注释描述了此行为的原因：

# Workers are created as daemon threads. This is done to allow the interpreter
# to exit when there are still idle threads in a ThreadPoolExecutor's thread
# pool (i.e. shutdown() was not called). However, allowing workers to die with
# the interpreter has two undesirable properties:
#   - The workers would still be running during interpreter shutdown,
#     meaning that they would fail in unpredictable ways.
#   - The workers could be killed while evaluating a work item, which could
#     be bad if the callable being evaluated has external side-effects e.g.
#     writing to a file.
#
# To work around this problem, an exit handler is installed which tells the
# workers to exit when their work queues are empty and then waits until the
# threads finish.

这是一个干净的出口的完整示例：

import asyncio
import multiprocessing
from multiprocessing import Queue
from concurrent.futures.thread import ThreadPoolExecutor
import sys


def main():
    executor = ThreadPoolExecutor()
    loop = asyncio.get_event_loop()
    # comment the following line and the shutdown will work smoothly
    fut = asyncio.ensure_future(print_some(executor))

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print("shutting down")
        queue.put(None)  # Poison pill
        loop.run_until_complete(fut)
        executor.shutdown()
        loop.stop()
        loop.close()


async def print_some(executor):
    print("Waiting...Hit CTRL+C to abort")
    loop = asyncio.get_event_loop()
    some = await loop.run_in_executor(executor, queue.get)
    print(some)


queue = None
if __name__ == '__main__':
    queue = Queue()
    main()

The run_until_complete(fut) call is needed to avoid a warning about a pending task hanging around when the asyncio eventloop exits. If you don't care about that, you can leave that call out.