《Python编程实战》8.3 定时任务、幂等与死信处理

import asyncio

async def run_every(interval: float, job, ticks: int):
    """每隔 interval 秒执行一次 job，共执行 ticks 次"""
    for i in range(ticks):
        await asyncio.sleep(interval)          # 固定间隔
        await job(i)

async def heartbeat(i):
    print(f"  [t={i}] 心跳任务执行")

async def main():
    await run_every(0.05, heartbeat, 3)

asyncio.run(main())

  [t=0] 心跳任务执行
  [t=1] 心跳任务执行
  [t=2] 心跳任务执行

import json
import time
import fakeredis

redis = fakeredis.FakeRedis(decode_responses=True)
SCHED = "app:q:delayed"      # score = 计划执行的时间戳（秒）

def enqueue_at(task_id: str, delay_s: float) -> float:
    score = time.time() + delay_s
    redis.zadd(SCHED, {json.dumps({"id": task_id}): score})
    return score

def poll_due(now: float | None = None) -> list[dict]:
    """取出所有到点的任务（score <= now）"""
    now = now if now is not None else time.time()
    due = redis.zrangebyscore(SCHED, "-inf", now)
    if due:
        redis.zremrangebyscore(SCHED, "-inf", now)
    return [json.loads(x) for x in due]

enqueue_at("t1", 0.05)
enqueue_at("t2", 0.15)
print("入队后待调度:", redis.zcard(SCHED))
time.sleep(0.08)
print("t=0.08 到点:", poll_due())
time.sleep(0.1)
print("t=0.18 到点:", poll_due())
print("剩余待调度:", redis.zcard(SCHED))

入队后待调度: 2
t=0.08 到点: [{'id': 't1'}]
t=0.18 到点: [{'id': 't2'}]
剩余待调度: 0

def run_once(idem_key: str, fn):
    if not redis.set(f"app:idem:{idem_key}", "1", nx=True, ex=3600):
        return "已处理过，跳过"
    return fn()

calls = 0

def charge():
    global calls
    calls += 1
    return f"扣款成功 #{calls}"

print("第一次:", run_once("order-1001-charge", charge))
print("重复提交:", run_once("order-1001-charge", charge))
print("实际扣款次数:", calls)

第一次: 扣款成功 #1
重复提交: 已处理过，跳过
实际扣款次数: 1

def process_payment(order_id: str, amount: int):
    key = f"app:idem:pay:{order_id}"
    if redis.set(key, json.dumps({"status": "processing"}), nx=True, ex=3600):
        result = {"order": order_id, "charged": amount}   # 真正扣款
        redis.set(key, json.dumps({"status": "done", "result": result}), ex=3600)
        return result, "首次执行"
    state = json.loads(redis.get(key))
    if state["status"] == "done":
        return state["result"], "重复请求，返回缓存结果"
    return None, "处理中，请稍后"

print(process_payment("O-1", 99))
print(process_payment("O-1", 99))
print(process_payment("O-1", 99))

({'order': 'O-1', 'charged': 99}, '首次执行')
({'order': 'O-1', 'charged': 99}, '重复请求，返回缓存结果')
({'order': 'O-1', 'charged': 99}, '重复请求，返回缓存结果')

DLQ = "app:q:default:dead"
QUEUE = "app:q:default"

def list_dead(limit: int = 10) -> list[dict]:
    return [json.loads(x) for x in redis.lrange(DLQ, 0, limit - 1)]

def replay(task_id: str) -> bool:
    """把指定死信重新投回主队列，并从死信移除"""
    for raw in redis.lrange(DLQ, 0, -1):
        d = json.loads(raw)
        if d["task"]["id"] == task_id:
            task = d["task"]
            task["attempts"] = 0          # 重置重试计数
            redis.lrem(DLQ, 1, raw)
            redis.lpush(QUEUE, json.dumps(task, ensure_ascii=False))
            return True
    return False

for tid, reason in [("a1", "SMTP 拒收"), ("b2", "下游 503")]:
    redis.lpush(DLQ, json.dumps({"task": {"id": tid, "name": "send_email", "attempts": 3},
                                 "reason": reason}, ensure_ascii=False))

print("死信列表:")
for d in list_dead():
    print("  ", d["task"]["id"], "->", d["reason"])
print("重放 a1:", replay("a1"))
print("重放后 主队列:", redis.llen(QUEUE), " 死信剩余:", redis.llen(DLQ))

死信列表:
   b2 -> 下游 503
   a1 -> SMTP 拒收
重放 a1: True
重放后 主队列: 1  死信剩余: 1