如果在装饰器中，不能使用 ProcessPoolExecutor？答案

【问题标题】：Cannot use ProcessPoolExecutor if in a decorator?如果在装饰器中，不能使用 ProcessPoolExecutor？
【发布时间】：2020-12-07 22:16:00
【问题描述】：

我有这个最小的例子：

from functools import wraps
from concurrent import futures
import random

def decorator(func):
    num_process = 4

    def impl(*args, **kwargs):
        with futures.ProcessPoolExecutor() as executor:
            fs = []
            for i in range(num_process):
                fut = executor.submit(func, *args, **kwargs)
                fs.append(fut)
            result = []
            for f in futures.as_completed(fs):
                result.append(f.result())
        return result
    return impl

@decorator
def get_random_int():
    return random.randint(0, 100)


if __name__ == "__main__":
    result = get_random_int()
    print(result)

如果我们尝试运行这个函数，我想我们会遇到以下错误：

_pickle.PicklingError: Can't pickle <function get_random_int at 0x7f06cee666a8>: it's not the same object as __main__.get_random_int

我认为这里的主要问题是“包装”装饰器本身会改变 func 对象，因此无法腌制。我觉得这很奇怪。我只是想知道是否有任何方法可以解决这种行为？如果可能的话，我想使用wraps。谢谢！

【问题讨论】：

同意。我见过类似的问题。它抱怨酸洗。想知道是否有人有办法解决这个问题。

标签： python python-3.x pickle concurrent.futures

【解决方案1】：

这是因为 run_in_executor 正在对装饰函数调用 functools.partial，请参阅：https://docs.python.org/3/library/asyncio-eventloop.html#asyncio-pass-keywords ~~部分对象的可腌制性参差不齐（请参阅：Are partial functions "officially" picklable?）但是~~请在此处查看此评论Pickling wrapped partial functions 部分函数仅在被腌制的函数位于全局名称空间中时才可腌制。我们知道 run_in_executor 和 ProcessPoolExecutor 将适用于非包装函数，因为该模式记录在 asyncio 中。为了解决这个问题，我装饰了一个虚拟函数并将我想要在多个进程中执行的函数作为参数传递给装饰器

from functools import wraps
from concurrent import futures
import random

def decorator(multiprocess_func):
    def _decorate(func):
        num_process = 4

        def impl(*args, **kwargs):
            with futures.ProcessPoolExecutor() as executor:
                fs = []
                for i in range(num_process):
                    fut = executor.submit(multiprocess_func, *args, **kwargs)
                    fs.append(fut)
                result = []
                for f in futures.as_completed(fs):
                    result.append(f.result())
            return result
        return impl
    return _decorate

def _get_random_int():
    return random.randint(0, 100)

@decorator(_get_random_int)
def get_random_int():
    return _get_random_int()


if __name__ == "__main__":
    result = get_random_int()
    print(result)

我最终决定不使用装饰器更干净

from concurrent import futures
import random

def decorator(multiprocess_func):
    num_process = 4

    def impl(*args, **kwargs):
        with futures.ProcessPoolExecutor(max_workers=num_process) as executor:
            fs = []
            for i in range(num_process):
                fut = executor.submit(multiprocess_func, *args, **kwargs)
                fs.append(fut)
            result = []
            for f in futures.as_completed(fs):
                result.append(f.result())
        return result
    return impl

def _get_random_int():
    return random.randint(0, 100)

get_random_int = decorator(_get_random_int)


if __name__ == "__main__":
    result = get_random_int()
    print(result)

类似于上面关于酸洗包装的部分函数的链接答案。

【讨论】：