如何向ThreadPoolExecutor的多线程传递同一非全局锁

你遇到的问题是executor.map要求后续参数必须是可迭代对象，直接传入lock会因为它不可迭代报错。以下是几种可行的解决方案：

原错误代码 #

import threading
from concurrent.futures import ThreadPoolExecutor

def main():
    lock = threading.Lock()
    
    with ThreadPoolExecutor(max_workers=10) as executor:
        executor.map(thread_function, range(10), lock)

def thread_function(number: int, lock: threading.Lock):
    with lock:
        print("Hello from the thread with argument: ", number)

if __name__ == '__main__':
    main()

解决方案1：使用itertools.repeat生成重复的lock引用 #

itertools.repeat可以生成一个重复返回同一元素的可迭代对象，指定重复次数后，刚好和range(10)的元素数量匹配，每个线程都会拿到同一个lock实例。

import threading
from concurrent.futures import ThreadPoolExecutor
from itertools import repeat

def main():
    lock = threading.Lock()
    
    with ThreadPoolExecutor(max_workers=10) as executor:
        # 生成包含10个相同lock引用的可迭代对象
        executor.map(thread_function, range(10), repeat(lock, 10))

def thread_function(number: int, lock: threading.Lock):
    with lock:
        print("Hello from the thread with argument: ", number)

if __name__ == '__main__':
    main()

解决方案2：用Lambda闭包传递lock #

通过lambda表达式包装目标函数，将lock作为闭包变量传入，这样不需要修改原函数的参数结构。

import threading
from concurrent.futures import ThreadPoolExecutor

def main():
    lock = threading.Lock()
    
    with ThreadPoolExecutor(max_workers=10) as executor:
        executor.map(lambda num: thread_function(num, lock), range(10))

def thread_function(number: int, lock: threading.Lock):
    with lock:
        print("Hello from the thread with argument: ", number)

if __name__ == '__main__':
    main()

解决方案3：用functools.partial绑定参数 #

使用partial预先将lock绑定到目标函数的参数上，生成一个只需要传入number的新函数，适配executor.map的参数要求。

import threading
from concurrent.futures import ThreadPoolExecutor
from functools import partial

def main():
    lock = threading.Lock()
    # 绑定lock到thread_function的lock参数
    bound_thread_func = partial(thread_function, lock=lock)
    
    with ThreadPoolExecutor(max_workers=10) as executor:
        executor.map(bound_thread_func, range(10))

def thread_function(number: int, lock: threading.Lock):
    with lock:
        print("Hello from the thread with argument: ", number)

if __name__ == '__main__':
    main()

原理说明 #

Python的函数参数传递是引用传递，以上三种方法本质都是让每个线程拿到同一个threading.Lock实例的引用，从而实现多线程间的同步。executor.map会按顺序从每个可迭代对象中取元素，组合后传递给目标函数，因此只要保证第二个参数是包含相同lock引用的可迭代对象即可。

内容的提问来源于stack exchange，提问作者glades