RabbitMQ集群实现消息的高可用性和负载均衡

# 在 node1, node2, node3 上分别执行
sudo systemctl start rabbitmq-server
# 或 rabbitmq-server -detached

# 在 node2 上执行
rabbitmqctl stop_app
rabbitmqctl reset # 如果是新节点，可不reset。如果是已存数据的节点，reset会清除数据！
rabbitmqctl join_cluster rabbit@node1 # 注意：rabbit是默认的Erlang节点名前缀
rabbitmqctl start_app
# 在 node3 上执行相同操作
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# 在任意节点执行
rabbitmqctl cluster_status

Cluster status of node rabbit@node1 ...
Basics
Cluster name: rabbit@node1
Disk Nodes
rabbit@node1
rabbit@node2
rabbit@node3
Running Nodes
rabbit@node1
rabbit@node2
rabbit@node3
...

# 设置一个策略，将名称以“mirrored.”开头的队列镜像到所有节点
rabbitmqctl set_policy ha-all "^mirrored\." '{"ha-mode":"all"}'
# 更常见的生产配置：镜像到多数节点（N/2+1），例如3节点集群中镜像到2个
rabbitmqctl set_policy ha-majority "^ha\." '{"ha-mode":"exactly", "ha-params":2, "ha-sync-mode":"automatic"}'

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ClusterConnectionExample {
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        // 关键：设置集群节点地址数组
        // 格式：amqp:// 可省略，默认端口5672
        Address[] addresses = new Address[] {
            new Address("node1", 5672),
            new Address("node2", 5672),
            new Address("node3", 5672)
        };
        // 自动重试连接。也可以使用外部负载均衡器（如HAProxy、F5）
        Connection connection = factory.newConnection(addresses, "MyApp-Client");
        System.out.println("成功连接到集群: " + connection.getAddress().getHostAddress());
        // ... 使用 connection 创建 Channel 进行后续操作
        connection.close();
    }
}

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class RabbitMQClusterConfig {
    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        // 设置集群节点地址
        factory.setAddresses("node1:5672,node2:5672,node3:5672");
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        // 开启Publisher Confirms，保证消息可靠投递
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 开启Publisher Returns，监听不可路由消息
        factory.setPublisherReturns(true);
        return factory;
    }
    @Bean
    public RabbitAdmin rabbitAdmin(CachingConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }
    @Bean
    public Queue mirroredQueue() {
        Map<String, Object> args = new HashMap<>();
        // 在代码中也可以声明队列参数，但通常更推荐用Policy在服务端统一定义
        // args.put("x-ha-policy", "all"); // 旧参数，已弃用
        // 推荐在RabbitMQ管理后台或通过rabbitmqctl set_policy设置
        return new Queue("order.queue", 
                         true,  // durable 持久化
                         false, // exclusive 非独占
                         false, // autoDelete 不自动删除
                         args);
    }
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange("order.exchange", true, false);
    }
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(mirroredQueue())
                             .to(orderExchange())
                             .with("order.created");
    }
}

[生产者/消费者] 
        |
[负载均衡器: HAProxy/Nginx] (Virtual IP)
        |
    [RabbitMQ集群]
    /      |      \
Node1    Node2    Node3 (磁盘节点，同机房)

listen rabbitmq_cluster
    bind 0.0.0.0:5670
    mode tcp
    balance roundrobin
    timeout client 3h
    timeout server 3h
    option clitcpka
    server node1 node1:5672 check inter 5s rise 2 fall 3
    server node2 node2:5672 check inter 5s rise 2 fall 3
    server node3 node3:5672 check inter 5s rise 2 fall 3

import org.springframework.amqp.core.QueueBuilder;
@Bean
public Queue quorumOrderQueue() {
    return QueueBuilder.durable("quorum.order.queue")
            .quorum() // 设置为仲裁队列
            .build();
}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import java.nio.charset.StandardCharsets;
public class OrderProducer {
    private final static String EXCHANGE_NAME = "order.exchange.ha";
    private final static String ROUTING_KEY = "order.created";
    private final static String QUEUE_NAME = "order.queue.ha";
    public static void main(String[] argv) throws Exception {
        // 1. 连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // 生产环境应为负载均衡器地址
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        // 2. 创建连接和信道
        // 使用try-with-resources确保资源关闭
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 3. 声明持久化的直连交换机和持久化的队列
            // 注意：这些定义会在集群节点间同步（元数据）
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
            // 队列声明，假设服务端已通过Policy将此队列设为镜像队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
            // 4. 开启Publisher Confirms (异步确认)
            channel.confirmSelect();
            // 5. 发送持久化消息
            String message = "订单消息: ORDER-20240426-001";
            channel.basicPublish(EXCHANGE_NAME, 
                                 ROUTING_KEY,
                                 MessageProperties.PERSISTENT_TEXT_PLAIN, // 关键：消息持久化
                                 message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] 发送 '" + message + "'");
            // 6. 等待Broker确认
            if (channel.waitForConfirms(5000)) {
                System.out.println(" [✓] 消息已得到Broker确认，投递成功。");
            } else {
                System.out.println(" [x] 消息未得到Broker确认，可能投递失败。");
                // 实际生产环境应实现重试或落库补偿逻辑
            }
        } // try-with-resources会自动关闭channel和connection
    }
}

import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
public class OrderConsumer {
    private final static String QUEUE_NAME = "order.queue.ha";
    public static void main(String[] argv) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 为了保证公平分发，在消费者端设置Qos
        int prefetchCount = 1; // 每次只预取1条消息
        channel.basicQos(prefetchCount);
        System.out.println(" [*] 等待消息。按 CTRL+C 退出");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] 收到 '" + message + "'");
            try {
                // 模拟业务处理耗时
                Thread.sleep(1000);
                System.out.println(" [✓] 业务处理完成: " + message);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                // 关键：手动确认消息，保证可靠性
                // deliveryTag, multiple
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                System.out.println(" [✓] 消息已手动确认。");
            }
        };
        // 消费消息，关闭自动确认(autoAck=false)，采用手动确认
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {
            System.out.println(" [x] 消费者被取消: " + consumerTag);
        });
    }
}