import redis
import json
from functools import wraps
r = redis.Redis(host='localhost', port=6379, db=0)
def idempotent(key_func, expire=300):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
生成幂等 key，例如基于消息体中的 order_id
        msg = kwargs.get('message') or args[0] if args else None
        if not msg:
            raise ValueError("message not found in args/kwargs")

        key = key_func(msg)
        if not key:
            raise ValueError("idempotent key is empty")

        # 尝试加锁（SET if not exists + expire）
        ok = r.set(key, "1", nx=True, ex=expire)
        if not ok:
            print(f"[SKIP] message already processed: {key}")
            return None

        try:
            result = func(*args, **kwargs)
            return result
        except Exception as e:
            # 可选：记录异常但不重试，避免死循环
            print(f"[ERROR] processing {key}: {e}")
            raise
    return wrapper
return decorator
使用示例
@idempotent(key_func=lambda msg: f"consume:{json.loads(msg.value()).get('order_id')}", expire=600)
def process_order_message(message):
data = json.loads(message.value())
order_id = data["order_id"]
执行真实业务逻辑：创建订单、扣减库存等
print(f"Processing order {order_id}")