在 ThreadPoolExecutor (TPE) 中，是否始终保证回调与提交的函数在同一线程中运行？

例如，我使用以下代码对此进行了测试。我运行了很多次，似乎 func 和 callback 总是在同一个线程中运行。

import concurrent.futures 
import random 
import threading 
import time 

executor = concurrent.futures.ThreadPoolExecutor(max_workers=3) 

def func(x): 
    time.sleep(random.random()) 
    return threading.current_thread().name 

def callback(future): 
    time.sleep(random.random()) 
    x = future.result() 
    cur_thread = threading.current_thread().name 
    if (cur_thread != x): 
        print(cur_thread, x) 

print('main thread: %s' % threading.current_thread()) 
for i in range(10000): 
    future = executor.submit(func, i) 
    future.add_done_callback(callback) 

但是，当我删除 time.sleep(random.random()) 语句时，它似乎失败了，即至少有几个 func 函数和 回调 没有在同一线程中运行。

对于我正在处理的项目，回调必须始终与提交的函数在同一线程上运行，因此我想确保 TPE 能够保证这一点。 (而且没有随机 sleep 的测试结果似乎令人费解)。

我查看了 source code for executors在运行回调之前，我们似乎没有将线程切换到主线程。但只是想确定一下。

最佳答案

文档不保证在哪个线程回调中运行。The only documented guarantee是回调将在属于添加回调的进程的线程中运行，但它可以是任何线程，因为您使用的是 ThreadPoolExecutor 而不是 ProcessPoolExecutor:

Added callables are called in the order that they were added and are always called in a thread belonging to the process that added them.

在当前的 ThreadPoolExecutor 实现中，执行回调的线程取决于添加回调时 Future 的状态，以及是否 Future取消。这些是实现细节；您不应该依赖它们，因为它们在不同的 Python 实现或不同版本中可能有所不同，并且它们如有更改，恕不另行通知。

如果您在 Future 完成后添加回调，回调将在您调用 add_done_callback 的任何线程中执行。您可以通过查看 看到这一点add_done_callback 来源:

def add_done_callback(self, fn):
    """Attaches a callable that will be called when the future finishes.

    Args:
        fn: A callable that will be called with this future as its only
            argument when the future completes or is cancelled. The callable
            will always be called by a thread in the same process in which
            it was added. If the future has already completed or been
            cancelled then the callable will be called immediately. These
            callables are called in the order that they were added.
    """
    with self._condition:
        if self._state not in [CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED]:
            self._done_callbacks.append(fn)
            return
    fn(self)

如果 Future 的状态指示它已取消或完成，则 fn 会立即在当前执行线程中调用。否则，它会添加到内部回调列表中，以便在 Future 完成时运行。

例如:

>>> def func(*args):
...  time.sleep(5)
...  print("func {}".format(threading.current_thread()))
>>> def cb(a): print("cb {}".format(threading.current_thread()))
... 
>>> fut = ex.submit(func)
>>> func <Thread(Thread-1, started daemon 140084551563008)>
>>> fut = e.add_done_callback(cb)
cb <_MainThread(MainThread, started 140084622018368)>

如果一个 future 被成功的 cancel 调用取消，那么执行取消的线程会立即调用所有回调:

def cancel(self):
    """Cancel the future if possible.
    Returns True if the future was cancelled, False otherwise. A future
    cannot be cancelled if it is running or has already completed.
    """
    with self._condition:
        if self._state in [RUNNING, FINISHED]:
            return False

        if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:
            return True

        self._state = CANCELLED
        self._condition.notify_all()

    self._invoke_callbacks()
    return True

否则，回调将由执行 future 任务的线程调用。

关于Python ThreadPoolExecutor - 是否保证回调与提交的函数在同一线程中运行？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/26021526/