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);
        });
    }
}