基于Python与HTTP Request实现多线程函数间交互的方案咨询

核心思路 #

不用手动加锁，用**线程安全的队列（queue.Queue）**做线程间通信更简单可靠：

• 函数A（HTTP服务）作为「生产者」，收到POST请求后往队列里丢创建文件的任务
• 函数B（循环线程）作为「消费者」，一直监听队列，拿到任务就创建文件，再把文件路径丢回队列给A
• 用Flask快速搭建HTTP接口，新手友好

步骤与代码 #

1. 先安装依赖：

pip install flask

2. 完整代码示例：

from flask import Flask, request
import threading
import queue
import uuid

# 线程安全队列，用于A和B之间传递任务/结果
task_queue = queue.Queue()
app = Flask(__name__)

def function_b():
    """一直循环监听队列，处理创建文件的任务"""
    while True:
        # 阻塞等待任务，直到队列有内容
        task = task_queue.get()
        if task == "create_file":
            # 用UUID生成唯一文件名，避免重复冲突
            file_name = f"result_{uuid.uuid4().hex}.txt"
            # 创建空文件
            with open(file_name, 'w', encoding='utf-8') as f:
                pass
            # 把生成的文件路径放回队列，传给A
            task_queue.put(file_name)
        # 标记当前任务处理完成
        task_queue.task_done()

@app.route('/create-file', methods=['POST'])
def function_a():
    """处理POST请求，触发文件创建并写入内容"""
    # 往队列发送创建文件的指令
    task_queue.put("create_file")
    # 等待B完成创建，获取文件路径
    file_path = task_queue.get()
    # 写入"hello world"
    try:
        with open(file_path, 'a', encoding='utf-8') as f:
            f.write("hello world")
        task_queue.task_done()
        return f"操作完成，文件路径：{file_path}", 200
    except Exception as e:
        task_queue.task_done()
        return f"写入失败：{str(e)}", 500

if __name__ == '__main__':
    # 启动B线程，设为守护线程，主程序退出时自动终止
    b_thread = threading.Thread(target=function_b, daemon=True)
    b_thread.start()
    # 启动Flask开发服务器，生产环境建议换用gunicorn等稳定服务器
    app.run(host='0.0.0.0', port=5000, debug=False)

关键细节说明 #

• 线程安全队列：queue.Queue内部已经实现了锁机制，不用自己手动加threading.Lock，避免了死锁等问题，比手动锁更省心
• 守护线程：把B线程设为守护线程，当Flask服务停止时，B会自动退出，不用手动处理线程终止
• 唯一文件名：用UUID生成文件名，避免多个请求同时触发时文件被覆盖，比用时间戳更可靠
• 异常处理：示例里加了简单的异常捕获，实际使用可以根据需求扩展（比如检查文件权限、磁盘空间等）

测试方法 #

启动程序后，用curl或者Postman发送POST请求：

curl -X POST http://localhost:5000/create-file

返回结果里会包含文件路径，打开文件就能看到"hello world"。

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