步骤 1：创建新的 4.2.x 实例（绿色集群）
cat << 'EOF' | kubectl -n <namespace> create -f -
apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
name: <green-instance-name>
spec:
replicas: 3
version: "4.2"
resources:
limits:
cpu: 1
memory: 2Gi
requests:
cpu: 1
memory: 2Gi
persistence:
storage: 1Gi
storageClassName: <storage-class>
service:
type: ClusterIP
rabbitmq:
additionalPlugins:
- rabbitmq_federation
- rabbitmq_federation_management
additionalConfig: |-
channel_max=1000
default_vhost=/
handshake_timeout=10000
heartbeat=60
max_message_size=134217728
queue_index_embed_msgs_below=4096
raft.wal_max_size_bytes=64000000
vm_memory_calculation_strategy=allocated
vm_memory_high_watermark.relative=0.4
EOF
等待绿色实例变为 Active：
kubectl -n <namespace> get rabbitmqclusters <green-instance-name> -w
步骤 2：从蓝色实例导出定义并导入到绿色实例
# Export definitions from blue
kubectl exec -n <namespace> <blue-pod-0> -- \
rabbitmqctl export_definitions /tmp/definitions.json
# Copy to local
kubectl cp <namespace>/<blue-pod-0>:/tmp/definitions.json ./definitions.json
# Copy to green pod
kubectl cp ./definitions.json <namespace>/<green-pod-0>:/tmp/definitions.json
# Import definitions into green
kubectl exec -n <namespace> <green-pod-0> -- \
rabbitmqctl import_definitions /tmp/definitions.json
验证定义是否已正确导入：
# Check exchanges, queues, and bindings on green
kubectl exec -n <namespace> <green-pod-0> -- rabbitmqctl list_exchanges name type
kubectl exec -n <namespace> <green-pod-0> -- rabbitmqctl list_queues name
kubectl exec -n <namespace> <green-pod-0> -- rabbitmqctl list_bindings
步骤 3：在绿色集群上配置 Federation 上游
先获取蓝色集群的内部服务地址：
kubectl -n <namespace> get svc -l app.kubernetes.io/name=<blue-instance-name>
然后获取蓝色集群的凭据：
# Get the default user secret name
kubectl -n <namespace> get rabbitmqclusters <blue-instance-name> \
-o jsonpath='{.status.defaultUser.secretReference.name}'
# Retrieve username and password
BLUE_USER=$(kubectl -n <namespace> get secret <blue-secret-name> \
-o jsonpath='{.data.username}' | base64 -d)
BLUE_PASS=$(kubectl -n <namespace> get secret <blue-secret-name> \
-o jsonpath='{.data.password}' | base64 -d)
echo "Blue cluster credentials: $BLUE_USER / $BLUE_PASS"
URL 编码
如果您的用户名或密码包含特殊字符（例如 @、:、/），请确保它们在下面的连接 URI 中已进行 URL 编码。
在 绿色 集群上配置指向 蓝色 的上游：
kubectl exec -n <namespace> <green-pod-0> -- \
rabbitmqctl set_parameter federation-upstream blue \
'{"uri":"amqp://<blue-user>:<blue-password>@<blue-service-name>.<namespace>.svc.cluster.local"}'
设置一个 Federation 策略以匹配所有队列：
kubectl exec -n <namespace> <green-pod-0> -- \
rabbitmqctl set_policy --apply-to queues blue-federation ".*" \
'{"federation-upstream":"blue"}'
策略优先级说明
在 RabbitMQ 中，如果多个策略匹配同一个队列，则只会应用优先级最高的那个。如果您导入的定义中已经包含覆盖这些队列的策略，上面的全局 blue-federation 策略可能不会生效（或者如果其优先级更高，可能会覆盖现有规则）。
在这种情况下，您应该手动更新绿色集群上的现有策略，在保留当前定义的同时，添加 "federation-upstream": "blue" 键。
验证 Federation 状态：
kubectl exec -n <namespace> <green-pod-0> -- rabbitmqctl eval \
'rabbit_federation_status:status().'
步骤 4：将消费者迁移到绿色集群
将消费者应用配置更新为指向绿色集群的服务地址。此时：
- 生产者仍然向 蓝色 集群发布消息。
- 消费者从 绿色 集群读取消息。
- Federation 会自动将消息从蓝色拉取到绿色。
您可以使用以下命令获取绿色集群的服务地址：
# Get the green cluster's service address
kubectl -n <namespace> get svc -l app.kubernetes.io/name=<green-instance-name>
验证消息是否正在被转发：
# Check that consumers on green are receiving messages
kubectl exec -n <namespace> <green-pod-0> -- \
rabbitmqctl list_queues name messages consumers
# Check federation link status
kubectl exec -n <namespace> <green-pod-0> -- \
rabbitmqctl eval 'rabbit_federation_status:status().'
重要
在将消费者从蓝色集群迁移出去之前，请确保绿色集群上的 Federation 链路处于 running 状态。如果链路未运行，消息将不会传输。
步骤 5：清空蓝色集群中的消息
当所有消费者都迁移到绿色实例后，Federation 会清空蓝色实例中的剩余消息。监控进度：
# Watch queue message counts on blue decrease to 0
kubectl exec -n <namespace> <blue-pod-0> -- \
rabbitmqctl list_queues name messages
如果积压较多，您也可以选择使用 Shovel 插件来加速清空：
# Enable shovel on green if not already enabled
kubectl exec -n <namespace> <green-pod-0> -- rabbitmq-plugins enable rabbitmq_shovel
# Create a shovel for a specific queue with large backlog
kubectl exec -n <namespace> <green-pod-0> -- \
rabbitmqctl set_parameter shovel shovel-blue-queue1 \
'{"src-protocol":"amqp091","src-uri":"amqp://<blue-user>:<blue-password>@<blue-service-name>.<namespace>.svc.cluster.local","src-queue":"<queue-name>","dest-protocol":"amqp091","dest-uri":"amqp://","dest-queue":"<queue-name>"}'
说明
结合 Federation 同时使用 Shovel 会并行传输消息，这可能导致消息到达顺序发生变化。如果消息顺序非常重要，请仅依赖 Federation。
步骤 6：将生产者迁移到绿色集群
当蓝色集群上的队列接近为空时：
-
停止所有生产者。
-
等待剩余消息通过 Federation 完全清空。
-
验证蓝色集群上的所有队列都为空：
kubectl exec -n <namespace> <blue-pod-0> -- \
rabbitmqctl list_queues name messages
-
更新生产者配置以指向绿色集群，并重启它们。
步骤 7：清理并下线蓝色集群
从绿色集群中移除 Federation 策略和上游配置：
kubectl exec -n <namespace> <green-pod-0> -- \
rabbitmqctl clear_policy blue-federation
kubectl exec -n <namespace> <green-pod-0> -- \
rabbitmqctl clear_parameter federation-upstream blue
删除蓝色集群：
kubectl -n <namespace> delete rabbitmqcluster <blue-instance-name>
