灰度发布方案

本文介绍RabbitMQ消费者灰度发布方案，包括流量切换、版本兼容与风险回滚。

定义

灰度发布是在保证服务不中断的前提下，将新版本消费者逐步接入消息队列，通过流量比例控制与指标监控验证新版本稳定性，异常时快速回滚的发布策略。

原理

灰度架构

                    ┌─────────────┐
                    │  Producer   │
                    └──────┬──────┘
                           │ publish
                           ▼
                    ┌─────────────┐
                    │  Exchange   │
                    └──────┬──────┘
                           │ binding
              ┌────────────┼────────────┐
              ▼            ▼            ▼
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │ Queue_v1 │ │ Queue_v2 │ │ DLQ      │
        │ (旧消费) │ │ (新消费) │ │ (死信)   │
        └────┬─────┘ └────┬─────┘ └──────────┘
             │             │
             ▼             ▼
        ┌──────────┐ ┌──────────┐
        │Consumer_A│ │Consumer_B│
        │ (v1.0)   │ │ (v1.1)   │
        └──────────┘ └──────────┘

流量切换策略

版本兼容要求

消息格式兼容:
├─ 字段向后兼容：新增字段有默认值
├─ 序列化协议兼容：JSON/Protobuf版本协商
└─ 路由键兼容：新旧消费者监听同Queue或不同Queue

示例

Maven依赖

<dependencies>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.20.0</version>
    </dependency>
</dependencies>

灰度消费者实现

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;

public class CanaryConsumer {
    private final String queue;
    private final String version;
    private final ConnectionFactory factory;

    public CanaryConsumer(String queue, String version) {
        this.queue = queue;
        this.version = version;
        this.factory = new ConnectionFactory();
        this.factory.setHost("localhost");
        this.factory.setPort(5672);
    }

    public void start(CountDownLatch latch) throws Exception {
        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            ch.basicQos(10); // 灰度期保守prefetch
            ch.queueDeclare(queue, true, false, false, null);

            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                try {
                    String body = new String(delivery.getBody());
                    processWithVersion(body, version);
                    ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                } catch (Exception e) {
                    // 灰度期异常记录但不reject，避免死信风暴
                    System.err.println("[" + version + "] Error: " + e.getMessage());
                }
            };

            ch.basicConsume(queue, false, deliverCallback, consumerTag -> {});
            latch.await(); // 保持运行
        }
    }

    private void processWithVersion(String message, String version) {
        // 版本差异化处理逻辑
        System.out.println("[" + version + "] Processing: " + message);
    }

    public static void main(String[] args) throws Exception {
        CountDownLatch latch = new CountDownLatch(1);

        // 启动旧版本消费者（95%流量）
        new Thread(() -> {
            try {
                new CanaryConsumer("order_queue", "v1.0").start(latch);
            } catch (Exception e) { e.printStackTrace(); }
        }).start();

        // 启动新版本消费者（5%流量）
        new Thread(() -> {
            try {
                new CanaryConsumer("order_queue", "v1.1").start(latch);
            } catch (Exception e) { e.printStackTrace(); }
        }).start();

        // 灰度控制：运行15分钟后评估
        Thread.sleep(15 * 60 * 1000);
        System.out.println("Canary period ended, evaluate metrics...");
        latch.countDown();
    }
}

流量切换控制

import com.rabbitmq.client.*;
import java.util.Map;
import java.util.HashMap;

public class TrafficController {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            // 通过Policy控制流量分配
            Map<String, Object> argsMap = new HashMap<>();
            argsMap.put("x-consumer-version", "v1.1");
            argsMap.put("x-traffic-percent", 20); // 20%流量

            ch.queueDeclare("order_queue_v2", true, false, false, argsMap);

            // 绑定到同一Exchange
            ch.queueBind("order_queue_v2", "order_exchange", "order.created");

            System.out.println("Traffic switched to 20% for v1.1");
        }
    }
}

回滚操作

# 停止新版本消费者
pkill -f "CanaryConsumer.*v1.1"

# 清理灰度队列
rabbitmqadmin delete queue name=order_queue_v2

# 恢复旧版本全量消费
rabbitmqctl set_policy revert_traffic "order_queue" '{"consumer-version":"v1.0"}' --apply-to queues

注意事项

灰度期间新旧版本必须兼容同格式消息，否则会导致消息处理失败。

灰度消费者prefetch应保守设置，避免新版本异常时影响大量消息。

异常处理建议记录日志但不reject，避免灰度异常引发死信风暴。

流量切换通过Policy或独立Queue实现，推荐Queue隔离更安全可靠。

回滚流程必须提前演练，确保5分钟内可完成全量回滚。

要点总结

• 灰度发布通过流量比例控制逐步验证新版本稳定性
• 流量切换分观察期、放量期、稳定期、全量期四个阶段
• 消息格式必须向后兼容，否则灰度无法进行
• 灰度消费者异常处理应保守，避免死信风暴
• 回滚流程必须提前演练，确保快速恢复