如何以推送模式而非轮询获取订单TTL到期通知并自动取消订单

针对你的需求，以下几个推送式方案可以替代轮询，实现订单到期的实时触发处理，且适配数百级的订单规模：

1. Redis 键空间通知（Keyspace Notifications） #

如果你的订单状态已经存在Redis中，这个方案最直接，利用Redis的键到期事件触发取消逻辑。

配置步骤： #

• 开启Redis的键到期通知：在redis.conf中设置notify-keyspace-events Ex，或者运行时执行CONFIG SET notify-keyspace-events Ex（重启后失效，需持久化配置）。
• 订阅__keyevent@<db>__:expired频道，监听键到期事件。

代码示例（Python）： #

import redis
import threading

def handle_expired_order(message):
    # 解析到期的键，提取订单ID（假设键格式为order:{order_id}）
    order_key = message['data'].decode('utf-8')
    order_id = order_key.split(':')[1]
    
    # 执行取消订单的幂等逻辑：更新订单状态、释放库存等
    print(f"自动取消到期订单：{order_id}")

# 初始化Redis连接
r = redis.Redis(host='localhost', port=6379, db=0)
# 订阅到期事件（db为0则用__keyevent@0__:expired）
pubsub = r.pubsub()
pubsub.subscribe(**{'__keyevent@0__:expired': handle_expired_order})

# 启动后台线程监听事件
listener_thread = threading.Thread(target=pubsub.run_in_thread, daemon=True)
listener_thread.start()

# 存储订单时设置TTL（例如15分钟=900秒）
order_id = "ORD12345"
r.setex(f'order:{order_id}', 900, "待支付")

注意事项： #

• Redis的到期通知是至少一次投递，可能出现重复通知，所以取消订单的逻辑必须实现幂等（比如先检查订单状态再执行取消）。
• 数百级订单的规模下，完全不会对Redis性能造成压力。

2. RabbitMQ 死信交换机（DLX）实现延迟队列 #

借助RabbitMQ的死信机制，将订单消息放入带TTL的普通队列，到期后消息自动转入死信队列，消费死信队列即可触发取消操作。

配置步骤： #

1. 声明死信交换机和死信队列，并绑定两者。
2. 声明普通业务队列，指定其死信交换机为步骤1的交换机。
3. 发送订单消息时设置TTL（消息级或队列级，消息级更灵活）。
4. 消费死信队列的消息，执行取消订单逻辑。

代码示例（Python + Pika）： #

import pika
import threading

# 连接RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 1. 声明死信交换机和队列
DLX_EXCHANGE = 'order_dlx'
DLX_QUEUE = 'order_dlx_queue'
channel.exchange_declare(exchange=DLX_EXCHANGE, exchange_type='direct')
channel.queue_declare(queue=DLX_QUEUE)
channel.queue_bind(exchange=DLX_EXCHANGE, queue=DLX_QUEUE, routing_key='order_expire')

# 2. 声明普通业务队列，绑定死信交换机
NORMAL_QUEUE = 'order_pending_queue'
channel.queue_declare(
    queue=NORMAL_QUEUE,
    arguments={
        'x-dead-letter-exchange': DLX_EXCHANGE,
        'x-dead-letter-routing-key': 'order_expire'
    }
)

# 3. 发送带TTL的订单消息
def send_order_delay_message(order_id, expire_seconds=900):
    channel.basic_publish(
        exchange='',
        routing_key=NORMAL_QUEUE,
        body=order_id.encode('utf-8'),
        properties=pika.BasicProperties(
            expiration=str(expire_seconds * 1000)  # TTL单位为毫秒
        )
    )
    print(f"订单 {order_id} 已加入延迟队列，将在{expire_seconds}秒后到期")

# 4. 消费死信队列，处理取消订单
def handle_expired_order(ch, method, properties, body):
    order_id = body.decode('utf-8')
    # 执行幂等的取消订单逻辑
    print(f"处理到期订单：取消 {order_id}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue=DLX_QUEUE, on_message_callback=handle_expired_order)

# 启动后台消费线程
consume_thread = threading.Thread(target=channel.start_consuming, daemon=True)
consume_thread.start()

# 测试发送订单
send_order_delay_message("ORD12345", 900)

注意事项： #

• 支持消息持久化，即使RabbitMQ重启，未到期的消息不会丢失。
• 同样需要保证取消逻辑的幂等性。

3. APScheduler 精准定时任务 #

如果你的系统没有引入Redis或RabbitMQ，直接用Python的APScheduler框架给每个订单设置到期触发的定时任务，适合数百级的小订单规模。

代码示例： #

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import time

# 配置任务持久化（避免服务重启丢失任务，可选）
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
scheduler = BackgroundScheduler(jobstores=jobstores)
scheduler.start()

def cancel_order(order_id):
    # 执行幂等的取消订单逻辑
    print(f"自动取消订单：{order_id}")

# 创建订单时，添加到期定时任务
order_id = "ORD12345"
expire_time = time.time() + 900  # 15分钟后到期
scheduler.add_job(
    cancel_order,
    'date',
    run_date=expire_time,
    args=[order_id],
    id=f"order_{order_id}",  # 设置唯一ID，避免重复添加
    replace_existing=True
)

# 保持程序运行（实际项目中可结合Web框架的启动逻辑）
try:
    while True:
        time.sleep(1)
except (KeyboardInterrupt, SystemExit):
    scheduler.shutdown()

注意事项： #

• 无需额外中间件，代码实现简单。
• 配置任务持久化后，服务重启不会丢失未执行的任务。
• 数百级任务对系统资源几乎无压力。

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