如何避免独立输入输出队列场景下的竞态条件问题

你担心的竞态条件必然会出现。
当前实现的输入队列、输出队列是两个独立的FIFO队列，没有任何机制将调用方的请求和返回的结果做绑定。多线程并发调用process函数时，线程往输入队列提交任务的顺序，和各个线程从输出队列取结果的顺序无法保证一一对应，就会出现线程拿到其他线程处理结果的问题。

有三种常用的无竞态实现方案，你可以根据场景选择：

方案1：绑定专属输出队列 #

每个process调用创建一个仅属于当前请求的单元素输出队列，和数据一起送入输入队列，工作线程处理完成后将结果写入对应的专属队列，调用方仅从自己的专属队列取结果，完全避免拿错的可能。
示例代码：

import queue

def process(item):
    # 创建当前请求专属的输出队列
    my_out_queue = queue.Queue(maxsize=1)
    inp_queue.put((item, my_out_queue))
    # 仅从自己的队列取结果
    return my_out_queue.get()

def separate_thread():
    while True:
        item, out_queue = inp_queue.get()
        processed_item = do_work(item)
        out_queue.put(processed_item)

方案2：用Future对象关联请求和结果 #

借助Python标准库concurrent.futures的Future对象实现请求和结果的绑定，还能天然支持异常传递，使用更简洁：
示例代码：

import queue
from concurrent.futures import Future

def process(item):
    fut = Future()
    inp_queue.put((item, fut))
    # 阻塞等待结果返回，异常也会同步抛出到调用方
    return fut.result()

def separate_thread():
    while True:
        item, fut = inp_queue.get()
        try:
            processed_item = do_work(item)
            fut.set_result(processed_item)
        except Exception as e:
            # 处理过程的异常会抛回调用process的线程
            fut.set_exception(e)

方案3：批量任务场景加请求唯一标识 #

如果是高频率小任务场景，创建大量临时队列/Future开销较高，可以给每个请求加唯一ID，输出队列返回时携带ID，调用方匹配自己的ID再消费结果：
示例代码：

import queue
import uuid
import threading

# 存储每个线程等待的请求ID
thread_local = threading.local()

def process(item):
    req_id = uuid.uuid4().hex
    thread_local.req_id = req_id
    inp_queue.put((req_id, item))
    while True:
        res_id, res = out_queue.get()
        if res_id == req_id:
            return res
        # 不是自己的结果放回队列，等待其他线程消费
        out_queue.put((res_id, res))

def separate_thread():
    while True:
        req_id, item = inp_queue.get()
        processed_item = do_work(item)
        out_queue.put((req_id, processed_item))

注意：该方案仅适合并发量不高的场景，高并发下反复放回队列会产生额外性能开销。

内容的提问来源于stack exchange，提问作者Nikolai K.