<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  rabbitmq:
    addresses: rabbitmq.example.com:5672  # 域名+端口（多節(jié)點(diǎn)用逗號分隔）
    username: admin
    password: your_password
    virtual-host: /
    # 連接超時設(shè)置
    connection-timeout: 30000  # 連接超時30秒
    # 心跳檢測（關(guān)鍵：檢測連接存活）
    requested-heartbeat: 60  # 心跳間隔60秒，避免被防火墻/負(fù)載均衡器斷開
    # 自動恢復(fù)配置（客戶端內(nèi)置重連機(jī)制）
    publisher-confirm-type: CORRELATED  # 確保消息發(fā)布確認(rèn)可靠性
    publisher-returns: true  # 開啟消息返回機(jī)制
    listener:
      simple:
        retry:
          enabled: true  # 開啟消費(fèi)者重試（避免連接抖動導(dǎo)致消息丟失）
          max-attempts: 3  # 最大重試次數(shù)
          initial-interval: 1000  # 重試初始間隔1秒
    # 連接工廠高級配置（通過Java代碼進(jìn)一步定制）

java -Dsun.net.inetaddr.ttl=10 -Dsun.net.inetaddr.negative.ttl=5 -jar your-app.jar

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.rabbitmq.client.Connection;
import java.net.InetAddress;
import java.net.UnknownHostException;
@Configuration
public class RabbitMQConfig {
    @Value("${spring.rabbitmq.addresses}")
    private String addresses;  // 原始配置的域名地址（如 rabbitmq.example.com:5672）
    @Value("${spring.rabbitmq.username}")
    private String username;
    @Value("${spring.rabbitmq.password}")
    private String password;
    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;
    @Bean
    public ConnectionFactory connectionFactory() {
        // 1. 解析域名獲取最新IP（核心：繞過DNS緩存）
        String resolvedAddresses = resolveDomainToIp(addresses);
        // 2. 創(chuàng)建RabbitMQ連接工廠
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setAddresses(resolvedAddresses);  // 使用解析后的IP地址
        factory.setUsername(username);
        factory.setPassword(password);
        factory.setVirtualHost(virtualHost);
        factory.setConnectionTimeout(30000);  // 連接超時30秒
        factory.setRequestedHeartbeat(60);  // 心跳間隔60秒
        // 3. 啟用自動恢復(fù)（關(guān)鍵：連接斷開后自動重建）
        factory.setAutomaticRecoveryEnabled(true);  // 開啟自動恢復(fù)
        factory.setNetworkRecoveryInterval(5000);  // 網(wǎng)絡(luò)恢復(fù)重試間隔5秒
        factory.setTopologyRecoveryEnabled(true);  // 恢復(fù)隊列/交換機(jī)綁定（拓?fù)浠謴?fù)）
        // 4. 注冊連接監(jiān)聽器（監(jiān)控連接狀態(tài)，觸發(fā)自定義重連）
        factory.addConnectionListener(new ConnectionListener() {
            @Override
            public void onClose(Connection connection) {
                // 連接關(guān)閉時觸發(fā)（非主動關(guān)閉）
                if (!connection.isOpen()) {
                    System.out.println("RabbitMQ連接已關(guān)閉，準(zhǔn)備重連...");
                    // 可在此處添加額外重連邏輯（如刷新IP后重建連接）
                    refreshConnection(factory);
                }
            }
            @Override
            public void onShutDown(ShutdownSignalException signal) {
                // 處理服務(wù)端主動關(guān)閉信號
                if (!signal.isInitiatedByApplication()) {
                    System.err.println("RabbitMQ連接被強(qiáng)制關(guān)閉，原因：" + signal.getReason());
                    refreshConnection(factory);
                }
            }
        });
        return factory;
    }
    // 解析域名獲取最新IP（替換域名中的主機(jī)為IP）
    private String resolveDomainToIp(String addresses) {
        try {
            // 分割多節(jié)點(diǎn)地址（如 "host1:5672,host2:5672"）
            String[] addressArray = addresses.split(",");
            StringBuilder resolved = new StringBuilder();
            for (String addr : addressArray) {
                String[] hostPort = addr.split(":");
                String host = hostPort[0];
                String port = hostPort.length > 1 ? hostPort[1] : "5672";
                // 解析域名獲取最新IP
                InetAddress[] inetAddresses = InetAddress.getAllByName(host);
                String latestIp = inetAddresses[0].getHostAddress();  // 取第一個IP（多IP可輪詢）
                resolved.append(latestIp).append(":").append(port).append(",");
            }
            // 移除最后一個逗號
            return resolved.substring(0, resolved.length() - 1);
        } catch (UnknownHostException e) {
            System.err.println("域名解析失敗，使用原始地址：" + e.getMessage());
            return addresses;  // 解析失敗時 fallback 到原始域名
        }
    }
    // 刷新連接（重建連接工廠）
    private void refreshConnection(CachingConnectionFactory factory) {
        try {
            // 1. 關(guān)閉舊連接
            factory.destroy();
            // 2. 重新解析域名獲取新IP
            String newAddresses = resolveDomainToIp(addresses);
            // 3. 更新連接工廠地址并重建連接
            factory.setAddresses(newAddresses);
            factory.createConnection();  // 主動創(chuàng)建新連接
            System.out.println("RabbitMQ重連成功，新地址：" + newAddresses);
        } catch (Exception e) {
            System.err.println("重連失敗，5秒后重試：" + e.getMessage());
            // 延遲重試（避免頻繁失敗）
            try {
                Thread.sleep(5000);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
            refreshConnection(factory);  // 遞歸重試
        }
    }
}

spring:
  rabbitmq:
    addresses: rabbitmq-primary.example.com:5672,rabbitmq-secondary.example.com:5672  # 主備順序

// 從服務(wù)發(fā)現(xiàn)獲取健康節(jié)點(diǎn)（偽代碼）
private List<String> getHealthyNodes() {
    // 調(diào)用服務(wù)發(fā)現(xiàn)API，獲取健康的RabbitMQ節(jié)點(diǎn)（如 "10.0.1.1:5672,10.0.2.1:5672"）
    return serviceDiscovery.getHealthyInstances("rabbitmq-service");
}

// 重連成功后校驗(yàn)隊列是否存在（示例）
@Autowired
private RabbitAdmin rabbitAdmin;
public void validateQueue(String queueName) {
    if (!rabbitAdmin.getQueueProperties(queueName).containsKey("queue")) {
        // 隊列不存在，重新聲明
        Queue queue = QueueBuilder.durable(queueName)
                .withArgument("x-message-ttl", 28800000)
                .build();
        rabbitAdmin.declareQueue(queue);
    }
}

management:
  endpoints:
    web:
      exposure:
        include: rabbithealth,health
  endpoint:
    rabbithealth:
      enabled: true  # 開啟RabbitMQ專屬健康檢查