为何我的Python多线程程序仅实际使用单个线程？

我编写的多线程程序未按预期运行，创建了nThreads个线程但实际仅使用其中一个。以下是最小可复现示例（MWE）：

import threading
import queue

NPerThread = 100
nThreads = 4

def worker(q: queue.Queue, oq: queue.Queue):
    while True:
        l = []
        threadIData = q.get(block=True)
        for i in range(threadIData["N"]):
            l.append(f"hello {i} from thread {threading.current_thread().name}")
        oq.put(l)
        q.task_done()

threadData = [{} for i in range(nThreads)]
inputQ = queue.Queue()
outputQ = queue.Queue()

for threadI in range(nThreads):
    threadData[threadI]["thread"] = threading.Thread(
        target=worker, args=(inputQ, outputQ),
        name=f"WorkerThread{threadI}"
    )
    threadData[threadI]["N"] = NPerThread
    threadData[threadI]["thread"].setDaemon(True)
    threadData[threadI]["thread"].start()


for threadI in range(nThreads):
    # start and end are in units of 8 bytes.
    inputQ.put(threadData[threadI])

inputQ.join()

outData = [None] * nThreads
count = 0
while not outputQ.empty():
    outData[count] = outputQ.get()
    count += 1


for i in outData:
    assert len(i) == NPerThread
    print(len(i))
print(outData)

补充说明 #

通过性能分析发现该错误，性能分析截图如下：

错误原因分析 #

问题出在任务分配时机和线程调度特性：

• 所有线程启动完成后才批量向队列放入4个任务，线程完成一个任务后会立刻回到循环顶部执行q.get()，而Python线程调度在短任务场景下，可能还没来得及切换到其他空闲线程，就被同一个线程抢完了所有任务。
• 队列的get()操作是抢占式的，没有任务分配的公平性保证，短任务场景下单线程连续获取任务是常见的调度结果。

修复方案 #

调整任务放入队列的时机，启动一个线程就立刻放入对应任务，给调度器足够时间切换线程，让多个线程都能获取到任务：

import threading
import queue

NPerThread = 100
nThreads = 4

def worker(q: queue.Queue, oq: queue.Queue):
    while True:
        l = []
        threadIData = q.get(block=True)
        for i in range(threadIData["N"]):
            l.append(f"hello {i} from thread {threading.current_thread().name}")
        oq.put(l)
        q.task_done()

threadData = [{} for i in range(nThreads)]
inputQ = queue.Queue()
outputQ = queue.Queue()

# 启动线程后立即放入对应任务
for threadI in range(nThreads):
    threadData[threadI]["thread"] = threading.Thread(
        target=worker, args=(inputQ, outputQ),
        name=f"WorkerThread{threadI}"
    )
    threadData[threadI]["N"] = NPerThread
    threadData[threadI]["thread"].setDaemon(True)
    threadData[threadI]["thread"].start()
    # 关键修改：启动线程后立即添加任务到队列
    inputQ.put(threadData[threadI])

inputQ.join()

outData = [None] * nThreads
count = 0
while not outputQ.empty():
    outData[count] = outputQ.get()
    count += 1


for i in outData:
    assert len(i) == NPerThread
    print(len(i))
print(outData)

如果需要更严格的任务分配公平性，也可以使用queue.PriorityQueue或者自定义带线程ID的任务分配逻辑，但上述调整已经能解决当前问题。

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