首页 文章

ThreadPoolExecutor中的worker不是真正的守护进程

提问于
浏览
3

我无法弄清楚的是,虽然 ThreadPoolExecutor 使用守护进程工作者,但即使主线程退出,它们仍会运行 .

我可以在python3.6.4中提供一个最小的例子:

import concurrent.futures
import time


def fn():
    while True:
        time.sleep(5)
        print("Hello")


thread_pool = concurrent.futures.ThreadPoolExecutor()
thread_pool.submit(fn)
while True:
    time.sleep(1)
    print("Wow")

主线程和工作线程都是无限循环 . 因此,如果我使用 KeyboardInterrupt 来终止主线程,我希望整个程序也会终止 . 但实际上工作线程仍在运行,即使它是一个守护程序线程 .

ThreadPoolExecutor 的源代码确认工作线程是守护程序线程:

t = threading.Thread(target=_worker,
                     args=(weakref.ref(self, weakref_cb),
                           self._work_queue))
t.daemon = True
t.start()
self._threads.add(t)

此外,如果我手动创建一个守护程序线程,它就像一个魅力:

from threading import Thread
import time


def fn():
    while True:
        time.sleep(5)
        print("Hello")


thread = Thread(target=fn)
thread.daemon = True
thread.start()
while True:
    time.sleep(1)
    print("Wow")

所以我真的无法弄清楚这种奇怪的行为 .

1 回答

  • 3

    突然......我找到了原因 . 根据 ThreadPoolExecutor 的更多源代码:

    # 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.
    
    _threads_queues = weakref.WeakKeyDictionary()
    _shutdown = False
    
    def _python_exit():
        global _shutdown
        _shutdown = True
        items = list(_threads_queues.items())
        for t, q in items:
            q.put(None)
        for t, q in items:
            t.join()
    
    atexit.register(_python_exit)
    

    有一个退出处理程序,将加入所有未完成的 Worker ......

相关问题