Python 中 ThreadPoolExecutor 上下文中的轻持久化

Question

我有一些 Python 代码使用 ThreadPoolExecutor 将昂贵的作业分流出来，我想跟踪其中哪些已经完成，这样如果我必须重新启动这个系统，我就不必重做已经完成的东西。在单线程上下文中，我可以只标记我在架子上所做的事情。下面是这个想法在多线程环境中的一个幼稚移植：

from concurrent.futures import ThreadPoolExecutor
import subprocess
import shelve


def do_thing(done, x):
    # Don't let the command run in the background; we want to be able to tell when it's done
    _ = subprocess.check_output(["some_expensive_command", x])
    done[x] = True


futs = []
with shelve.open("done") as done:
    with ThreadPoolExecutor(max_workers=18) as executor:
        for x in things_to_do:
            if done.get(x, False):
                continue
            futs.append(executor.submit(do_thing, done, x))
            # Can't run `done[x] = True` here--have to wait until do_thing finishes
        for future in futs:
            future.result()

    # Don't want to wait until here to mark stuff done, as the whole system might be killed at some point
    # before we get through all of things_to_do

我能摆脱这个吗？ documentation for shelve 不包含任何关于线程安全的保证，所以我认为没有。

那么处理这个问题的简单方法是什么？我认为也许坚持done[x] = True 到future.add_done_callback 会做到这一点，但that will often run in the same thread as the future itself。也许有一种锁定机制可以很好地与 ThreadPoolExecutor 配合使用？对我来说，编写一个休眠然后检查已完成期货的循环似乎更清晰。

Answer 1

虽然您仍在最外层的with 上下文管理器中，但done 搁置只是一个普通的python 对象——它仅在上下文管理器关闭并运行其__exit__ 方法时写入磁盘.因此，由于GIL（只要您使用的是 CPython），它与任何其他 python 对象一样是线程安全的。

具体来说，done[x] = True 的重新分配是线程安全的/将以原子方式完成。

需要注意的是，虽然搁置的__exit__ 方法将在 Ctrl-C 之后运行，但如果 python 进程突然结束则不会运行，并且搁置不会保存到磁盘。

为了防止此类故障，我建议使用基于文件的轻量级线程安全数据库，例如 sqllite3。