Python threading多线程下多函数动态承接列表元素处理的实现问题

解决方案 #

核心实现思路为使用Python标准库自带的线程安全队列queue.Queue存储待处理的数字，5个线程分别绑定fun_1到fun_5，每个线程循环从队列取数执行对应函数，直到队列无剩余任务为止，天然满足「先执行完的线程先取下一个未使用数字」的调度要求，无需手动实现锁和任务分配逻辑。

完整可运行代码 #

import threading
from random import randint
from time import sleep
from queue import Queue

# 原有业务代码无需修改
def main_fun(number):
    num = randint(1,10)
    sleep(num)
    return f"{num} --> {number}"

# 可选打印锁，避免多线程同时打印时内容错乱
print_lock = threading.Lock()
def fun_1(number):
    res = main_fun(number)
    with print_lock:
        print(f"fun_1  :  {res}")

def fun_2(number):
    res = main_fun(number)
    with print_lock:
        print(f"fun_2  :  {res}")

def fun_3(number):
    res = main_fun(number)
    with print_lock:
        print(f"fun_3  :  {res}")

def fun_4(number):
    res = main_fun(number)
    with print_lock:
        print(f"fun_4  :  {res}")

def fun_5(number):
    res = main_fun(number)
    with print_lock:
        print(f"fun_5  :  {res}")

a = [1,2,3,4,5,6,7,8,9,10]

# 调度逻辑代码
# 初始化队列，按顺序存入所有待处理数字
task_queue = Queue()
for num in a:
    task_queue.put(num)

# 线程工作逻辑：绑定目标函数，循环取队列任务执行
def worker(target_func):
    while not task_queue.empty():
        try:
            current_num = task_queue.get(block=True, timeout=1)
        except:
            break
        target_func(current_num)
        task_queue.task_done()

# 初始化5个线程，分别绑定对应fun_*函数
threads = [
    threading.Thread(target=worker, args=(fun_1,)),
    threading.Thread(target=worker, args=(fun_2,)),
    threading.Thread(target=worker, args=(fun_3,)),
    threading.Thread(target=worker, args=(fun_4,)),
    threading.Thread(target=worker, args=(fun_5,)),
]

# 启动所有线程
for t in threads:
    t.start()

# 等待所有任务处理完成
task_queue.join()
# 等待所有线程退出
for t in threads:
    t.join()

print("所有数字处理完成")

逻辑说明 #

• 队列采用先进先出规则，前5个取出的数字必然是1、2、3、4、5，完全匹配初始阶段的启动要求
• 队列本身为线程安全实现，同一时间只会有一个线程取到同一个数字，不会出现重复分配、漏分配的问题
• 线程执行完当前任务后会自动取下一个队列中剩余的数字，符合调度规则

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