Pika长耗时任务结束后调用ack方法抛异常问题排查

1. 首先解释报错里的>Q：这是Python struct模块的格式化字符串，>代表使用大端字节序，Q代表要打包的值为8字节无符号长整型，要求传入参数必须是整数类型，这和抛出的required argument is not an integer报错直接对应——说明Pika内部编码ACK帧时拿到的delivery_tag不是合法整数。
2. 核心根因是函数参数绑定错位：
   • 你定义的__ack_message方法形参顺序为(self, delivery_tag, ack)，但在do_work中构造functools.partial时，错误地把Channel对象ch作为第一个预绑定参数传入，后面才跟真正的整型delivery_tag。
   • 当回调被Pika事件循环触发时，参数对应关系完全错乱：形参delivery_tag实际接收到的是Channel对象，形参ack实际接收到的是原本的整型delivery_tag。
   • 后续调用self.channel.basic_ack(delivery_tag)时，传入的是Channel实例而非整数，Pika执行struct.pack('>Q', self.delivery_tag)时自然抛出类型错误。
3. 附带逻辑缺陷：do_work中没有对长耗时任务做异常捕获，也没有传入ack参数对应的布尔值，即使参数顺序正确，也会出现参数缺失、异常时消息无法nack的问题。

按以下步骤调整代码即可解决问题：

• 修正__ack_message的参数传递逻辑，去掉多余的Channel传参，保证整型delivery_tag被正确传入
• 在长耗时任务外层加异常捕获，根据执行结果生成ack/nack对应的标记位
• 调用basic_ack/basic_nack时使用关键字参数传参，避免后续再出现位置参数错位问题

修复后的核心代码如下：

def __ack_message(self, delivery_tag, ack):
    print("Inside the __ack_message function")
    if not self.channel.is_open:
        return
    try:
        if ack:
            # 显式指定关键字参数，避免位置错位
            self.channel.basic_ack(delivery_tag=delivery_tag)
        else:
            self.channel.basic_nack(delivery_tag=delivery_tag, requeue=False)
    except Exception as e:
        print(f"Failure inside __ack_message consumer.py report: {e}", "error")
        # 异常时兜底尝试nack
        try:
            self.channel.basic_nack(delivery_tag=delivery_tag, requeue=False)
        except:
            pass

def do_work(self, ch, delivery_tag, body):
    thread_id = threading.get_ident()
    print(f'Thread id:{thread_id} Delivery tag: {delivery_tag} Message body: {body}')
    ack_flag = False
    try:
        print("I am inside the message_consume")
        message = None
        if body != b'':
            message = json.loads(body)
        # 执行长耗时处理任务
        self.slave_object.push_to_job_queue(self.connection, message)
        ack_flag = True
    except Exception as e:
        print(f"Process message error: {str(e)}")
        ack_flag = False
    finally:
        # 构造回调时仅预绑定正确的delivery_tag和ack标记，不再传入多余的ch参数
        cb = functools.partial(self.__ack_message, delivery_tag, ack_flag)
        ch.connection.add_callback_threadsafe(cb)

def on_message(self, ch, method_frame, _header_frame, body, args):
    print("Inside on_message function")
    thrds = args
    delivery_tag = method_frame.delivery_tag
    t = threading.Thread(target=self.do_work, args=(ch, delivery_tag, body))
    t.start()
    thrds.append(t)

补充注意点：

• Pika的连接、Channel实例不是线程安全的，所有涉及Channel/Connection的操作（ack/nack、发布消息等）都必须通过add_callback_threadsafe投递到Pika所属的事件循环线程执行，你当前的线程模型是符合要求的，只需要修正参数错误即可。
• 建议定期清理thrds列表中已经执行完成的线程，避免线程对象一直占用内存。

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