故障转移机制

RabbitMQ 集群节点故障时，通过队列镜像与连接重定向实现自动故障转移。

故障转移架构

核心组件

客户端 → 负载均衡器 → [Node A, Node B, Node C]
                              ↓ 故障
                        [Node B, Node C] 继续服务

故障转移层次：

1. 连接层：客户端重连到其他健康节点
2. 队列层：镜像队列在其他节点恢复
3. 元数据层：Mnesia 在其他节点保留完整副本

队列镜像故障转移

镜像队列原理

镜像队列将队列内容复制到集群多个节点：

主节点（Master）: 接收发布消息，同步到镜像节点
镜像节点（Mirror）: 接收同步数据，等待升级为主节点

同步模式：

• 同步模式：消息写入主节点后同步到所有镜像节点
• 异步模式：主节点先写入，后台异步同步

镜像队列配置

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

public class MirroredQueueExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            
            // 策略方式配置镜像队列（推荐）
            // rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}'
            
            ch.queueDeclare("ha.queue", true, false, false, null);
            
            // 发布消息到镜像队列
            ch.basicPublish("", "ha.queue",
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                "HA message".getBytes());
            
            System.out.println("消息已发布到镜像队列");
            System.out.println("主节点宕机后，镜像节点自动升级为主节点");
        }
    }
}

主节点故障转移

Node A (Master) ──同步──→ Node B (Mirror)
         ↓ 宕机
Node B 检测主节点离线 → 升级为新 Master → 继续服务

故障转移流程：

1. 镜像节点检测主节点心跳超时
2. 镜像节点选举新主节点
3. 新主节点接管队列读写
4. 元数据更新，指向新主节点
5. 消费者自动重连到新主节点

镜像队列故障转移期间可能有少量消息丢失（主节点已接收但未同步的消息）。

连接故障转移

客户端重连

import com.rabbitmq.client.*;
import java.util.Arrays;
import java.util.List;

public class ConnectionFailoverExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        
        // 配置多个节点地址
        List<Address> addresses = Arrays.asList(
            new Address("node1.example.com", 5672),
            new Address("node2.example.com", 5672),
            new Address("node3.example.com", 5672)
        );
        
        // 自动故障转移：按顺序尝试连接
        Connection conn = factory.newConnection(addresses);
        Channel ch = conn.createChannel();
        
        ch.queueDeclare("failover.queue", true, false, false, null);
        ch.basicPublish("", "failover.queue", null, "Test message".getBytes());
        
        System.out.println("连接已建立，当前节点: " + conn.getAddress());
        System.out.println("节点宕机时自动尝试下一个地址");
        
        ch.close();
        conn.close();
    }
}

自动重连配置

import com.rabbitmq.client.*;

public class AutoReconnectExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        
        // 启用自动恢复
        factory.setAutomaticRecoveryEnabled(true);
        // 设置恢复间隔（毫秒）
        factory.setNetworkRecoveryInterval(5000);
        
        Connection conn = factory.newConnection();
        Channel ch = conn.createChannel();
        
        // 启用自动恢复后，连接断开时自动重连
        // 拓扑恢复：重新声明交换器、队列、绑定
        // 连接恢复：重新建立 TCP 连接
        
        System.out.println("自动恢复已启用");
        System.out.println("网络中断后 5 秒自动尝试重连");
        
        ch.close();
        conn.close();
    }
}

脑裂规避

脑裂场景

网络分区导致集群分裂为多个子集群：

分区前: [A, B, C] 互通
分区后: [A, B] ──断开── [C]

脑裂问题：

1. 两个子集群各自认为自己是主集群
2. 队列出现多个主节点
3. 消息写入冲突，数据不一致

规避策略

推荐配置 pause_if_all_down：

# rabbitmq.conf
cluster_partition_handling = pause_if_all_down

该策略下：

1. 少数派节点检测到与多数派失联
2. 少数派节点暂停服务，不接受请求
3. 网络恢复后，少数派自动恢复
4. 确保只有一个子集群提供服务

pause_if_all_down 策略可完全避免脑裂，但少数派节点暂停期间服务不可用。

故障检测

心跳检测

RabbitMQ 使用 Erlang 分布式节点心跳检测：

Node A ──heartbeat──→ Node B
Node A ←──ack──────── Node B

检测机制：

1. 节点定期发送心跳消息（默认 60 秒）
2. 超时未收到响应，标记节点为 down
3. 集群元数据更新，移除该节点
4. 镜像队列在其他节点恢复

健康检查

import com.rabbitmq.client.*;

public class HealthCheckExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {
            
            // 通过 HTTP API 检查节点健康状态
            // GET http://localhost:15672/api/health/checks/alarms
            // 返回 {"status":"ok"} 表示节点健康
            
            // 也可以通过 basicPublish 测试队列可用性
            ch.queueDeclare("health.check", true, false, false, null);
            ch.basicPublish("", "health.check", null, "health check".getBytes());
            
            System.out.println("节点健康，队列可用性测试通过");
        }
    }
}

注意事项

镜像队列同步有性能开销，节点数越多同步延迟越大，建议镜像节点数 ≤ 2。

故障转移期间可能丢失未同步的消息，对数据一致性要求极高的场景应使用 Quorum Queues。

pause_if_all_down 策略下，少数派节点暂停服务期间无法自动恢复，需网络完全恢复。

客户端自动恢复仅恢复连接和拓扑，已发送但未确认的消息需业务层处理。

要点总结

• 故障转移分为连接层、队列层、元数据层三个层次
• 镜像队列将内容同步到多个节点，主节点宕机时镜像节点升级
• 客户端通过配置多节点地址实现连接自动故障转移
• 启用 AutomaticRecoveryEnabled 实现连接与拓扑自动恢复
• 脑裂规避推荐使用 pause_if_all_down 策略
• 镜像节点数建议 ≤ 2，避免同步延迟过大

文章存放路径：D:\git2\jwdev\articles\RABBITMQ\专家\高可用与容灾\故障转移机制.md