使用Pika的add_callback_threadsafe时程序停止后消息确认丢失的解决方法

你的问题核心在于：add_callback_threadsafe只是把消息确认任务提交到Pika的事件循环队列，但主线程在等待工作线程完成后直接关闭连接，此时Pika的事件循环可能已经停止（比如用BlockingConnection时，start_consuming退出后事件循环就停了），导致队列里的确认任务没机会被执行。

解决步骤 #

1. 跟踪待处理的确认任务数

用线程安全的计数器记录所有已提交但未执行的确认任务，每个确认任务执行完毕后递减计数器：

import threading

# 线程安全的计数器和锁
pending_ack_count = 0
count_lock = threading.Lock()

def confirm_ack(ch, delivery_tag):
    global pending_ack_count
    try:
        ch.basic_ack(delivery_tag=delivery_tag)
    finally:
        with count_lock:
            pending_ack_count -= 1

2. 提交确认任务时更新计数器

在工作线程中提交确认任务前，先增加计数器：

def process_batch(ch, batch_delivery_tags):
    global pending_ack_count
    # 批次业务处理逻辑...
    
    # 提交确认任务前更新计数器
    with count_lock:
        pending_ack_count += len(batch_delivery_tags)
    
    # 逐个提交确认任务
    for tag in batch_delivery_tags:
        ch.connection.add_callback_threadsafe(
            lambda tag=tag: confirm_ack(ch, tag)
        )

3. 优雅停止时等待所有确认任务执行完毕

在SIGINT处理函数中，等待工作线程完成后，手动驱动Pika的事件循环，直到计数器归0，再关闭资源：

def sigint_handler(signum, frame):
    print("收到停止信号，等待工作线程完成...")
    # 等待所有工作线程结束
    for thread in worker_threads:
        thread.join()
    
    print("处理剩余消息确认...")
    # 循环处理事件，直到所有确认完成
    while True:
        with count_lock:
            if pending_ack_count == 0:
                break
        # 处理事件，超时0.1秒避免死等
        connection.process_data_events(time_limit=0.1)
    
    # 关闭通道和连接
    channel.basic_cancel(consumer_tag='your_consumer_tag')
    channel.close()
    connection.close()
    exit(0)

关键注意事项 #

• 如果使用的是异步连接（如SelectConnection），需要确保事件循环继续运行直到所有确认任务完成，可通过IOLoop的add_future或同步事件来实现。
• 确认回调必须捕获所有异常，避免因单个确认失败导致计数器无法递减，进而卡住停止流程。
• 计数器操作必须加锁，防止多线程下的竞态问题。

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