如何配置Flask Webhook 实现前序请求处理完成后再处理下一个请求

要实现请求串行处理、避免重复创建草稿的需求，核心是对草稿操作逻辑加互斥锁，保证同一时间只有一个请求能执行「查草稿->创建/更新草稿」的完整流程，具体实现分两类场景：

单进程多线程部署场景（默认Flask启动场景） #

直接用Python标准库的线程锁即可，为了不影响不同收件人邮箱的请求处理效率，推荐按邮箱维度做细粒度锁，而非全局锁：

import threading
from flask import Flask, request

app = Flask(__name__)
# 存储每个邮箱对应的互斥锁
draft_locks = {}
# 全局锁，用于保证draft_locks字典的线程安全
global_lock = threading.Lock()

@app.route('/webhook', methods=['POST'])
def get_webhook():
    # some processing...
    # 假设此处已经获取到curator_email_address、song_url、artist、song_name参数
    
    # 先获取当前邮箱对应的锁
    with global_lock:
        if curator_email_address not in draft_locks:
            draft_locks[curator_email_address] = threading.Lock()
    curator_lock = draft_locks[curator_email_address]

    # 对同一个邮箱的草稿操作全程加锁，前一个请求处理完后一个才会进入
    with curator_lock:
        # Get drafts
        gmail_drafts = gmail.get_drafts()
        if gmail_drafts['resultSizeEstimate'] > 0:
            gmail_drafts = map(gmail.get_draft, [draft['id'] for draft in gmail_drafts['drafts']])
            drafts_to = {draft['id']:draft['message']['payload']['headers'][4]['value'] for draft in list(gmail_drafts)}

            if curator_email_address not in list(drafts_to.values()):
                create_draft(curator_email_address, song_url, artist, song_name)
            else:
                draft_id = list(drafts_to.keys())[0]
                update_draft(draft_id, song_url, artist, song_name)
        else:
            create_draft(curator_email_address, song_url, artist, song_name)

    return request.json, 200

多进程部署场景（uWSGI/Gunicorn多进程启动） #

线程锁在多进程场景下不生效，需要使用分布式锁，最常用的是基于Redis实现：

import redis
import time

redis_client = redis.Redis(host='localhost', port=6379, db=0)
LOCK_TIMEOUT = 30 # 锁超时时间，避免异常导致锁永久占用

@app.route('/webhook', methods=['POST'])
def get_webhook():
    # some processing...
    # 假设此处已经获取到curator_email_address、song_url、artist、song_name参数
    lock_key = f"draft_lock:{curator_email_address}"
    
    # 抢锁
    while not redis_client.set(lock_key, "locked", ex=LOCK_TIMEOUT, nx=True):
        time.sleep(0.1) # 抢不到就等待100ms再试
    
    try:
        # 原有查草稿、创建/更新草稿逻辑放此处
        gmail_drafts = gmail.get_drafts()
        # 其余原有逻辑不变，和单进程版本逻辑一致
    finally:
        # 处理完成主动释放锁
        redis_client.delete(lock_key)

    return request.json, 200

注意事项 #

• 锁的粒度要控制好，只锁「草稿校验+修改」的核心逻辑，不要把和草稿操作无关的预处理逻辑也放到锁里，避免不必要的性能损耗
• 分布式锁必须设置超时时间，防止请求异常中断导致锁无法释放，后续同邮箱的请求全部阻塞

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