如何在Python中并行运行两个函数？错误排查与正确实现

我尝试并行运行两段循环逻辑的代码但未成功，两段代码单独运行正常，总耗时约30分钟，最终都会返回字典列表，但并行时函数还是按顺序执行，还出现报错。

单独运行的代码示例 #

# 单独运行的代码示例
import time

def functionA(A, B):
    dictA = []
    for i in range(A, B):
        print(i, "from A")
        time.sleep(1)
        for p in range(0, 10):
            dictA.append({i:p})
    return dictA

def functionB(C, D):
    dictB = []
    for i in range(C, D):
        print(i, "from B")
        time.sleep(1)
        for p in range(0, 10):
            dictB.append({i:p})
    return dictB

DictA = functionA(0, 10)
DictB = functionB(10, 20)

尝试并行的错误代码 #

# 尝试并行但存在问题的代码
import threading

DictA = threading.Thread(target=functionA(0, 10))
DictA.start()               
DictB = threading.Thread(target=functionB(10, 20))
DictB.start()
# 后续需要在线程完成后执行的代码

运行后的错误输出 #

0 from A
1 from A
2 from A
3 from A
4 from A
5 from A
6 from A
7 from A
8 from A
9 from A
Exception in thread Thread-12:
Traceback (most recent call last):
  File "/home/maestro/.pyenv/versions/3.9.13/lib/python3.9/threading.py", line 980, in _bootstrap_inner
    self.run()
  File "/home/maestro/.pyenv/versions/3.9.13/lib/python3.9/threading.py", line 917, in run
    self._target(*self._args, **self._kwargs)
TypeError: 'list' object is not callable
10 from B
11 from B
12 from B
13 from B
14 from B
15 from B
16 from B
17 from B
18 from B
19 from B
Exception in thread Thread-13:
Traceback (most recent call last):
  File "/home/maestro/.pyenv/versions/3.9.13/lib/python3.9/threading.py", line 980, in _bootstrap_inner
    self.run()
  File "/home/maestro/.pyenv/versions/3.9.13/lib/python3.9/threading.py", line 917, in run
    self._target(*self._args, **self._kwargs)
TypeError: 'list' object is not callable

正确实现方案 #

错误原因分析 #

你创建Thread对象时犯了核心错误：直接调用functionA(0,10)并把结果传给target参数。这会导致函数在主线程中立即执行完毕，而不是交给子线程运行。Thread的target需要传入函数对象本身，不能是函数执行后的返回值（这里返回的是列表，所以报错'list' object is not callable）。另外，线程无法直接返回结果，需要通过队列或其他方式获取返回值。

方法1：使用threading.Thread + 队列获取返回值 #

通过队列传递线程的执行结果，手动管理线程生命周期：

import threading
import time
from queue import Queue

def functionA(A, B, queue):
    dictA = []
    for i in range(A, B):
        print(i, "from A")
        time.sleep(1)
        for p in range(0, 10):
            dictA.append({i:p})
    queue.put(dictA)

def functionB(C, D, queue):
    dictB = []
    for i in range(C, D):
        print(i, "from B")
        time.sleep(1)
        for p in range(0, 10):
            dictB.append({i:p})
    queue.put(dictB)

# 创建队列用于接收线程返回值
queue_a = Queue()
queue_b = Queue()

# 创建线程：target传函数对象，参数用args元组传入
thread_a = threading.Thread(target=functionA, args=(0, 10, queue_a))
thread_b = threading.Thread(target=functionB, args=(10, 20, queue_b))

# 启动线程
thread_a.start()
thread_b.start()

# 等待两个线程执行完毕，再继续后续代码
thread_a.join()
thread_b.join()

# 从队列中取出结果
DictA = queue_a.get()
DictB = queue_b.get()

# 后续可以正常使用DictA和DictB
print("并行执行完成，结果已获取")

方法2：使用concurrent.futures.ThreadPoolExecutor（更简洁） #

用线程池自动管理线程和结果，代码更简洁：

import time
from concurrent.futures import ThreadPoolExecutor

def functionA(A, B):
    dictA = []
    for i in range(A, B):
        print(i, "from A")
        time.sleep(1)
        for p in range(0, 10):
            dictA.append({i:p})
    return dictA

def functionB(C, D):
    dictB = []
    for i in range(C, D):
        print(i, "from B")
        time.sleep(1)
        for p in range(0, 10):
            dictB.append({i:p})
    return dictB

# 创建线程池，指定最大工作线程数为2
with ThreadPoolExecutor(max_workers=2) as executor:
    # 提交任务到线程池，返回Future对象
    future_a = executor.submit(functionA, 0, 10)
    future_b = executor.submit(functionB, 10, 20)
    
    # 通过Future对象获取执行结果
    DictA = future_a.result()
    DictB = future_b.result()

# 后续代码
print("并行执行完成，结果已获取")

关键注意点 #

1. Thread的target参数：必须传入函数对象（比如functionA），不能加括号调用（functionA(0,10)是执行函数并返回结果，不是函数对象）。
2. 参数传递：线程的参数通过args元组传递，比如args=(0,10,queue_a)。
3. 等待线程完成：用join()方法（方法1）或future.result()（方法2）确保后续代码在并行任务结束后执行。
4. 结果获取：线程无法直接返回结果，需要通过队列、全局变量或Future对象来传递结果，推荐用队列或ThreadPoolExecutor的方式，更安全可靠。

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