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

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

# 在任意節(jié)點(diǎn)執(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è)置一個(gè)策略，將名稱(chēng)以“mirrored.”開(kāi)頭的隊(duì)列鏡像到所有節(jié)點(diǎn)
rabbitmqctl set_policy ha-all "^mirrored\." '{"ha-mode":"all"}'
# 更常見(jiàn)的生產(chǎn)配置：鏡像到多數(shù)節(jié)點(diǎn)（N/2+1），例如3節(jié)點(diǎn)集群中鏡像到2個(gè)
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é)點(diǎn)地址數(shù)組
        // 格式：amqp:// 可省略，默認(rèn)端口5672
        Address[] addresses = new Address[] {
            new Address("node1", 5672),
            new Address("node2", 5672),
            new Address("node3", 5672)
        };
        // 自動(dòng)重試連接。也可以使用外部負(fù)載均衡器（如HAProxy、F5）
        Connection connection = factory.newConnection(addresses, "MyApp-Client");
        System.out.println("成功連接到集群: " + connection.getAddress().getHostAddress());
        // ... 使用 connection 創(chuàng)建 Channel 進(jìn)行后續(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é)點(diǎn)地址
        factory.setAddresses("node1:5672,node2:5672,node3:5672");
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        // 開(kāi)啟Publisher Confirms，保證消息可靠投遞
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 開(kāi)啟Publisher Returns，監(jiān)聽(tīng)不可路由消息
        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<>();
        // 在代碼中也可以聲明隊(duì)列參數(shù)，但通常更推薦用Policy在服務(wù)端統(tǒng)一定義
        // args.put("x-ha-policy", "all"); // 舊參數(shù)，已棄用
        // 推薦在RabbitMQ管理后臺(tái)或通過(guò)rabbitmqctl set_policy設(shè)置
        return new Queue("order.queue", 
                         true,  // durable 持久化
                         false, // exclusive 非獨(dú)占
                         false, // autoDelete 不自動(dòng)刪除
                         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)者/消費(fèi)者] 
        |
[負(fù)載均衡器: HAProxy/Nginx] (Virtual IP)
        |
    [RabbitMQ集群]
    /      |      \
Node1    Node2    Node3 (磁盤(pán)節(jié)點(diǎn)，同機(jī)房)

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è)置為仲裁隊(duì)列
            .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. 連接工廠(chǎng)
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // 生產(chǎn)環(huán)境應(yīng)為負(fù)載均衡器地址
        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. 聲明持久化的直連交換機(jī)和持久化的隊(duì)列
            // 注意：這些定義會(huì)在集群節(jié)點(diǎn)間同步（元數(shù)據(jù)）
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
            // 隊(duì)列聲明，假設(shè)服務(wù)端已通過(guò)Policy將此隊(duì)列設(shè)為鏡像隊(duì)列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
            // 4. 開(kāi)啟Publisher Confirms (異步確認(rèn))
            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確認(rèn)
            if (channel.waitForConfirms(5000)) {
                System.out.println(" [?] 消息已得到Broker確認(rèn)，投遞成功。");
            } else {
                System.out.println(" [x] 消息未得到Broker確認(rèn)，可能投遞失敗。");
                // 實(shí)際生產(chǎn)環(huán)境應(yīng)實(shí)現(xiàn)重試或落庫(kù)補(bǔ)償邏輯
            }
        } // try-with-resources會(huì)自動(dòng)關(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ā)，在消費(fèi)者端設(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ù)處理耗時(shí)
                Thread.sleep(1000);
                System.out.println(" [?] 業(yè)務(wù)處理完成: " + message);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                // 關(guān)鍵：手動(dòng)確認(rèn)消息，保證可靠性
                // deliveryTag, multiple
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                System.out.println(" [?] 消息已手動(dòng)確認(rèn)。");
            }
        };
        // 消費(fèi)消息，關(guān)閉自動(dòng)確認(rèn)(autoAck=false)，采用手動(dòng)確認(rèn)
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {
            System.out.println(" [x] 消費(fèi)者被取消: " + consumerTag);
        });
    }
}