RabbitMQ已关闭时rabbitpy发消息无异常及publisher_confirms死锁问题

先拆解你遇到的两个核心问题的根源，再给出针对性的解决方案：

问题根源分析 #

1. RabbitMQ断开后发布无异常提示 #

你使用了全局应用级的连接和channel，但eventlet是协程并发模型，多个协程会共享这个底层TCP连接。当RabbitMQ服务器关闭时，连接可能处于「半关闭」状态，rabbitpy默认不会主动检测连接存活状态，导致发送消息时无法触发预期异常——协程误以为socket还可写，但数据实际无法送达。

2. 启用channel.enable_publisher_confirms()后死循环 #

rabbitpy的publisher confirms是同步阻塞等待RabbitMQ的确认帧，但eventlet的monkey patch修改了标准库IO行为，rabbitpy的部分底层IO逻辑和协程调度不兼容，导致等待确认的代码永远收不到响应，陷入无限循环。

解决方案一：用连接池替代全局连接+前置健康检查 #

不要复用单个全局连接，改用连接池管理RabbitMQ连接，每个协程（请求）从池内获取连接/channel，且发布前先检查连接存活状态。

示例代码：

import rabbitpy
import json
from flask import Flask
import eventlet
eventlet.monkey_patch()

app = Flask(__name__)

# 实现简单的RabbitMQ连接池
class RabbitMQPool:
    def __init__(self, amqp_url, pool_size=5):
        self.amqp_url = amqp_url
        self.pool = []
        self.pool_size = pool_size
        self._init_pool()

    def _init_pool(self):
        for _ in range(self.pool_size):
            try:
                conn = rabbitpy.Connection(self.amqp_url, heartbeat=10)
                self.pool.append(conn)
            except Exception as e:
                print(f"初始化连接池失败: {str(e)}")

    def get_connection(self):
        # 从池内取连接，失效则丢弃重建
        while self.pool:
            conn = self.pool.pop()
            try:
                conn.send_heartbeat()  # 发送心跳检测连接
                return conn
            except rabbitpy.exceptions.ConnectionClosed:
                continue
        # 池内无可用连接，新建一个
        return rabbitpy.Connection(self.amqp_url, heartbeat=10)

    def return_connection(self, conn):
        if len(self.pool) < self.pool_size and conn.is_open:
            self.pool.append(conn)

# 初始化连接池
rabbit_pool = RabbitMQPool('amqp://guest:guest@127.0.0.1:5672/%2f')

@app.route('/publish')
def publish_message():
    conn = None
    try:
        conn = rabbit_pool.get_connection()
        channel = conn.channel()
        channel.enable_publisher_confirms()  # 安全启用确认机制
        
        message = rabbitpy.Message(channel, json.dumps({"test": "test"}))
        confirmed = message.publish('test', 'test', mandatory=True)
        
        if not confirmed:
            return "消息未被RabbitMQ确认", 500
        return "消息发布成功", 200
    except rabbitpy.exceptions.ConnectionClosed as e:
        return f"RabbitMQ服务器未运行/连接已关闭: {str(e)}", 503
    except Exception as e:
        return f"发布失败: {str(e)}", 500
    finally:
        if conn:
            rabbit_pool.return_connection(conn)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

解决方案二：适配eventlet协程处理Publisher Confirms #

如果必须使用全局连接，可通过eventlet的异步协程处理确认等待，避免阻塞主线程：

示例代码：

import rabbitpy
import json
from flask import Flask
import eventlet
eventlet.monkey_patch()

app = Flask(__name__)

# 初始化带心跳检测的全局连接
def init_rabbit_connection():
    conn = rabbitpy.Connection('amqp://guest:guest@127.0.0.1:5672/%2f', heartbeat=10)
    # 启动协程定期发送心跳，检测连接状态
    eventlet.spawn_n(_heartbeat_worker, conn)
    return conn

def _heartbeat_worker(conn):
    while conn.is_open:
        try:
            conn.send_heartbeat()
            eventlet.sleep(5)
        except rabbitpy.exceptions.ConnectionClosed:
            break

app.conn = init_rabbit_connection()
app.channel = app.conn.channel()
app.channel.enable_publisher_confirms()

@app.route('/publish')
def publish_message():
    try:
        # 检查连接是否存活，失效则重建
        if not app.conn.is_open:
            app.conn = init_rabbit_connection()
            app.channel = app.conn.channel()
            app.channel.enable_publisher_confirms()
        
        message = rabbitpy.Message(app.channel, json.dumps({"test": "test"}))
        # 用异步协程处理确认等待，避免死循环
        def _handle_confirm(msg):
            try:
                confirmed = msg.publish('test', 'test', mandatory=True)
                if not confirmed:
                    print("警告：消息未被RabbitMQ确认")
            except Exception as e:
                print(f"发布异常: {str(e)}")
        
        eventlet.spawn_n(_handle_confirm, message)
        return "消息已提交发布", 202
    except rabbitpy.exceptions.ConnectionClosed as e:
        return f"RabbitMQ服务器未运行: {str(e)}", 503
    except Exception as e:
        return f"发布失败: {str(e)}", 500

关键注意事项 #

1. 避免全局共享连接/channel：协程环境下共享socket易出现IO竞争和状态不一致，连接池是更可靠的方案。
2. 启用心跳检测：通过RabbitMQ心跳机制定期检测连接存活，一旦断开及时重建。
3. 异步处理Publisher Confirms：在eventlet协程中，不要同步等待确认，用异步协程处理避免阻塞死循环。

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