Psycopg2：事务成功提交后执行异步操作的可靠实现方案问询

这个问题太典型了——我在做分布式系统的时候踩过好几次类似的坑，分享几个经过生产环境验证的可靠方案：

1. 利用事务回调/钩子（最直接的方案） #

大部分ORM或数据库框架都提供了事务提交后的回调机制，这是最省心的做法。比如Java Spring里的TransactionSynchronizationManager、Python Django的post_commit信号，或者.NET的TransactionScope完成事件，都能帮你把操作绑定到事务成功的节点上。

举个Spring的代码例子：

// 在你的业务方法内部
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
    @Override
    public void afterCommit() {
        // 这里放心发RabbitMQ消息就行，只有事务提交成功才会执行
        rabbitTemplate.convertAndSend("order-exchange", "order.created", orderMessage);
    }
});

核心逻辑就是：把消息发送逻辑注册到事务的提交成功钩子里，事务回滚的话这个钩子根本不会触发，完美避免了提前发消息的问题。

2. 本地消息表（最终一致性兜底方案） #

如果你的框架不支持事务回调，或者需要应对服务宕机这类极端情况，本地消息表是个稳得一批的方案：

• 第一步：在同一个事务里，把要发送的消息内容插入到本地DB的pending_messages表（和你的业务操作同事务，要么一起成功要么一起回滚）
• 第二步：启动一个后台定时任务（比如Quartz、Celery Beat），定期扫描pending_messages里状态为待发送的消息
• 第三步：尝试发送消息到MQ，成功后把消息状态改成已发送；失败的话重试几次，实在不行标记为失败留待人工处理

这个方案的最大优势是容错性极强——哪怕事务提交后服务立刻宕机，重启后定时任务还能接着处理未发送的消息，绝不会丢消息。

3. MQ原生事务消息（依赖MQ能力的方案） #

如果你的MQ支持事务消息（比如RabbitMQ的事务模式、RocketMQ的分布式事务消息），可以直接用它的原生能力来做。

拿RabbitMQ举个Python的例子：

# 开启RabbitMQ事务
channel.tx_select()
try:
    # 先执行你的业务DB操作
    db.session.commit()
    # 发送消息到队列
    channel.basic_publish(exchange='order-exchange', routing_key='order.notify', body=message_json)
    # 提交MQ事务
    channel.tx_commit()
except Exception as e:
    # 任何一步出错，同时回滚DB和MQ事务
    db.session.rollback()
    channel.tx_rollback()
    raise e

不过要注意：RabbitMQ的事务模式会降低吞吐量，因为是同步阻塞的。如果系统对性能要求高，建议结合本地消息表来用，或者用publisher confirm模式替代。

几个必须注意的点 #

• 绝对别在事务提交前发消息：哪怕你觉得“马上就要提交了”，也可能因为DB锁冲突、网络波动等原因导致事务回滚，这时候已经发出去的消息就会造成数据不一致，排查起来超级麻烦。
• 一定要处理消息重复：不管用哪种方案，都可能出现消息重复发送的情况（比如MQ重试、定时任务重复扫描），所以你的消费者必须实现幂等性——比如根据消息ID判断是否已经处理过，避免重复执行业务逻辑。

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