# 在 node1, node2, node3 上分別執(zhí)行
sudo systemctl start rabbitmq-server
# 或 rabbitmq-server -detached

# 在 node2 上執(zhí)行
rabbitmqctl stop_app
rabbitmqctl reset # 如果是新節(jié)點，可不reset。如果是已存數(shù)據(jù)的節(jié)點，reset會清除數(shù)據(jù)！
rabbitmqctl join_cluster rabbit@node1 # 注意：rabbit是默認的Erlang節(jié)點名前綴
rabbitmqctl start_app
# 在 node3 上執(zhí)行相同操作
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# 在任意節(jié)點執(zhí)行
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
...

# 設(shè)置一個策略，將名稱以“mirrored.”開頭的隊列鏡像到所有節(jié)點
rabbitmqctl set_policy ha-all "^mirrored\." '{"ha-mode":"all"}'
# 更常見的生產(chǎn)配置：鏡像到多數(shù)節(jié)點（N/2+1），例如3節(jié)點集群中鏡像到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("/");
        // 關(guān)鍵：設(shè)置集群節(jié)點地址數(shù)組
        // 格式：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 創(chuàng)建 Channel 進行后續(xù)操作
        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();
        // 設(shè)置集群節(jié)點地址
        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，監(jiān)聽不可路由消息
        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<>();
        // 在代碼中也可以聲明隊列參數(shù)，但通常更推薦用Policy在服務(wù)端統(tǒng)一定義
        // args.put("x-ha-policy", "all"); // 舊參數(shù)，已棄用
        // 推薦在RabbitMQ管理后臺或通過rabbitmqctl set_policy設(shè)置
        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");
    }
}

[生產(chǎn)者/消費者] 
        |
[負載均衡器: HAProxy/Nginx] (Virtual IP)
        |
    [RabbitMQ集群]
    /      |      \
Node1    Node2    Node3 (磁盤節(jié)點，同機房)

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() // 設(shè)置為仲裁隊列
            .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"); // 生產(chǎn)環(huán)境應(yīng)為負載均衡器地址
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        // 2. 創(chuàng)建連接和信道
        // 使用try-with-resources確保資源關(guān)閉
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 3. 聲明持久化的直連交換機和持久化的隊列
            // 注意：這些定義會在集群節(jié)點間同步（元數(shù)據(jù)）
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
            // 隊列聲明，假設(shè)服務(wù)端已通過Policy將此隊列設(shè)為鏡像隊列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
            // 4. 開啟Publisher Confirms (異步確認)
            channel.confirmSelect();
            // 5. 發(fā)送持久化消息
            String message = "訂單消息: ORDER-20240426-001";
            channel.basicPublish(EXCHANGE_NAME, 
                                 ROUTING_KEY,
                                 MessageProperties.PERSISTENT_TEXT_PLAIN, // 關(guān)鍵：消息持久化
                                 message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] 發(fā)送 '" + message + "'");
            // 6. 等待Broker確認
            if (channel.waitForConfirms(5000)) {
                System.out.println(" [?] 消息已得到Broker確認，投遞成功。");
            } else {
                System.out.println(" [x] 消息未得到Broker確認，可能投遞失敗。");
                // 實際生產(chǎn)環(huán)境應(yīng)實現(xiàn)重試或落庫補償邏輯
            }
        } // try-with-resources會自動關(guān)閉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();
        // 為了保證公平分發(fā)，在消費者端設(shè)置Qos
        int prefetchCount = 1; // 每次只預(yù)取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 {
                // 模擬業(yè)務(wù)處理耗時
                Thread.sleep(1000);
                System.out.println(" [?] 業(yè)務(wù)處理完成: " + message);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                // 關(guān)鍵：手動確認消息，保證可靠性
                // deliveryTag, multiple
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                System.out.println(" [?] 消息已手動確認。");
            }
        };
        // 消費消息，關(guān)閉自動確認(autoAck=false)，采用手動確認
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {
            System.out.println(" [x] 消費者被取消: " + consumerTag);
        });
    }
}