Linux环境Python3服务优雅停机：RabbitMQ集群流量排空方案咨询

嘿，针对你这个多节点RabbitMQ消费者升级时要安全停机、排空流量的需求，我给你整理几个在Linux+Python3环境下实用的方案，都是生产环境里跑通的思路：

本质上我们要实现的是：收到停机信号后，停止接收新请求→处理完当前正在执行的业务→安全退出进程，结合RabbitMQ的消费特性和Python的信号处理能力就能搞定。

方案1：信号驱动的基础优雅停机 #

首先利用Python的signal模块捕获Linux系统的停机信号（比如SIGTERM，这是systemctl stop或默认kill命令发送的信号），通过线程安全的标志控制消费循环退出，同时确保现有消息处理完成。

完整代码示例 #

import signal
import time
import pika

# 用全局变量作为停止标志（单线程场景足够，多线程建议用threading.Event）
should_stop = False

def handle_shutdown_signal(signum, frame):
    global should_stop
    print(f"收到停止信号 {signum}，开始优雅停机...")
    should_stop = True

def main():
    # 注册信号处理器：处理SIGTERM（系统停机指令）和SIGINT（Ctrl+C）
    signal.signal(signal.SIGTERM, handle_shutdown_signal)
    signal.signal(signal.SIGINT, handle_shutdown_signal)

    # 初始化RabbitMQ连接和通道
    connection = pika.BlockingConnection(pika.ConnectionParameters('rabbitmq-cluster'))
    channel = connection.channel()
    channel.queue_declare(queue='request_queue')

    def callback(ch, method, properties, body):
        print(f"开始处理消息: {body.decode()}")
        # 模拟实际业务处理逻辑
        time.sleep(2)
        # 手动确认消息：必须等处理完成再确认，避免消息丢失
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print(f"消息处理完成: {body.decode()}")

    # 预取1条消息：避免一次性从队列拿太多消息，减少停机时待处理的任务量
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(queue='request_queue', on_message_callback=callback)

    print("服务启动成功，等待消息中...")
    # 循环消费，每1秒检查一次停止标志
    while not should_stop:
        try:
            # 使用process_data_events而非start_consuming，后者是永久阻塞无法检查标志
            channel.process_data_events(time_limit=1)
        except pika.exceptions.ConnectionClosedByBroker:
            break
        except pika.exceptions.AMQPChannelError as e:
            print(f"RabbitMQ通道错误: {e}")
            break
        except pika.exceptions.AMQPConnectionError as e:
            print(f"RabbitMQ连接断开，5秒后重试...: {e}")
            time.sleep(5)
            # 重连后重新初始化通道和消费配置
            connection = pika.BlockingConnection(pika.ConnectionParameters('rabbitmq-cluster'))
            channel = connection.channel()
            channel.queue_declare(queue='request_queue')
            channel.basic_qos(prefetch_count=1)
            channel.basic_consume(queue='request_queue', on_message_callback=callback)

    # 优雅收尾：关闭RabbitMQ通道和连接
    print("开始关闭RabbitMQ连接...")
    channel.close()
    connection.close()
    print("服务已安全停止")

if __name__ == "__main__":
    main()

关键细节说明 #

• 信号选择：一定要用SIGTERM或SIGINT，绝对不能用SIGKILL（kill -9），后者会直接强制杀死进程，完全无法执行优雅停机逻辑。
• 消息确认：必须开启手动确认basic_ack，如果用自动确认，消息一旦被取出就会被标记为已消费，停机时未处理完的消息会永久丢失。
• 消费循环控制：用process_data_events(time_limit=1)代替start_consuming()，让消费循环每1秒返回一次，及时检查停止标志，避免永久阻塞。

方案2：结合服务注册的进阶流量排空（适合多节点集群） #

如果你的集群有服务发现组件（比如Consul、Etcd）或者前端有负载均衡，建议在收到停机信号后先把节点从服务列表中注销，这样新的客户端请求会直接转发到其他正常节点，彻底实现流量排空，再处理完现有消息后退出。

代码片段示例（以Consul为例） #

修改信号处理函数即可：

import consul

def handle_shutdown_signal(signum, frame):
    global should_stop
    print(f"收到停止信号 {signum}，开始优雅停机...")
    # 第一步：从服务注册中心注销当前节点
    c = consul.Consul(host='your-consul-server')
    c.agent.service.deregister('your-server-service-name')
    print("节点已从注册中心注销，不再接收新请求")
    # 第二步：设置停止标志，开始处理剩余消息
    should_stop = True

生产环境配套配置 #

如果用systemd管理你的Python服务，记得在service配置文件中添加以下参数，确保系统给足时间让服务完成优雅停机：

[Unit]
Description=RabbitMQ Consumer Service
After=network.target rabbitmq-server.target

[Service]
User=your-user
ExecStart=/usr/bin/python3 /path/to/your/consumer.py
KillSignal=SIGTERM  # 指定发送SIGTERM信号而非默认的SIGTERM（其实默认就是，但明确写更稳妥）
TimeoutStopSec=300  # 给5分钟时间处理剩余消息，可根据业务调整
Restart=on-failure

[Install]
WantedBy=multi-user.target

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