连接心跳机制

RabbitMQ通过心跳机制检测连接活性，在空闲连接超时或网络中断时自动触发重连，避免连接假死。

定义

心跳是RabbitMQ客户端与服务端之间定期交换的控制帧。双方约定心跳间隔，若间隔内无数据交换，则发送Heartbeat帧。连续错过2次心跳即判定连接失效，触发断线重连。

心跳工作原理

客户端                服务端
  |--- Heartbeat ----->|
  |<-- Heartbeat ------|
  |                    |
  | (约定间隔内无数据)   |
  |--- Heartbeat ----->|  <- 检测连接活性
  |<-- Heartbeat ------|
  |                    |
  | (连续2次未收到)     |
  |--- 断开连接 --------|  <- 判定失效

心跳参数配置

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;

public class HeartbeatConfig {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");
        
        // 心跳间隔（秒），0表示禁用
        factory.setRequestedHeartbeat(60);
        
        // 连接超时（毫秒）
        factory.setConnectionTimeout(30000);
        
        // 握手超时（毫秒）
        factory.setHandshakeTimeout(10000);
        
        // 网络恢复间隔（毫秒）
        factory.setNetworkRecoveryInterval(5000);
        
        // 启用自动恢复
        factory.setAutomaticRecoveryEnabled(true);
        factory.setTopologyRecoveryEnabled(true);
        
        try (Connection connection = factory.newConnection()) {
            System.out.println("连接已建立，心跳间隔: 60秒");
        }
    }
}

心跳参数说明

自动重连配置

import com.rabbitmq.client.*;
import java.io.IOException;

public class AutoRecoveryExample {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        // 启用自动恢复
        factory.setAutomaticRecoveryEnabled(true);
        factory.setTopologyRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(5000);
        factory.setRequestedHeartbeat(30);
        
        // 添加连接监听器
        factory.addConnectionListener(new ConnectionListener() {
            @Override
            public void handleCreate(Connection connection) {
                System.out.println("连接已建立: " + connection.getAddress());
            }
            
            @Override
            public void handleFailure(Connection connection, Throwable cause) {
                System.err.println("连接失败: " + cause.getMessage());
            }
            
            @Override
            public void handleShutdown(Connection connection, ShutdownSignalReason reason) {
                System.out.println("连接关闭，原因: " + reason);
            }
        });
        
        Connection connection = factory.newConnection();
        System.out.println("已连接，等待自动重连测试...");
        
        // 模拟网络中断后恢复
        // 实际场景中RabbitMQ重启或网络抖动时会自动重连
    }
}

心跳超时处理

import com.rabbitmq.client.*;

public class HeartbeatTimeoutHandler {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setRequestedHeartbeat(30);
        factory.setAutomaticRecoveryEnabled(true);
        
        Connection connection = factory.newConnection();
        
        // 添加关闭钩子
        connection.addShutdownListener(cause -> {
            System.out.println("连接关闭，原因: " + cause.getReason());
            System.out.println("是否自动恢复: " + cause.isInitiatedByShutdownHook());
            
            if (cause.getReason() == ShutdownSignalReason.MISSED_HEARTBEAT) {
                System.err.println("心跳超时，连接将被自动重建");
            }
        });
        
        Channel channel = connection.createChannel();
        channel.queueDeclare("heartbeat_queue", true, false, false, null);
        
        // 保持程序运行
        Thread.sleep(Long.MAX_VALUE);
    }
}

心跳调优场景

场景1：高频率消息传输

// 消息频繁时无需心跳检测，可适当增大间隔
factory.setRequestedHeartbeat(120); // 2分钟

场景2：空闲连接保活

// 连接长时间空闲，需较短心跳防止NAT/防火墙断开
factory.setRequestedHeartbeat(30); // 30秒

场景3：禁用心跳（不推荐）

// 仅适用于极短时连接或本地测试
factory.setRequestedHeartbeat(0); // 禁用

注意事项

心跳间隔由客户端提议，服务端可拒绝并返回自己期望的值，最终取双方较大值。

连续2次未收到心跳帧即判定连接失效，实际超时时间约为 heartbeat * 2。

启用自动恢复后，RabbitMQ Java Client会自动重连并重建Topology（队列、交换器、绑定），需设置setTopologyRecoveryEnabled(true)。

心跳间隔不宜过短（<10秒），会增加服务器CPU负担；也不宜过长（>120秒），故障检测延迟大。

网络中间件（防火墙、NAT、负载均衡器）可能断开长时间空闲连接，心跳可起到保活作用。

心跳与连接池配合

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

public class HeartbeatWithPool {
    public static void main(String[] args) {
        GenericObjectPoolConfig<Connection> config = new GenericObjectPoolConfig<>();
        config.setTimeBetweenEvictionRunsMillis(30000); // 30秒检查一次空闲连接
        config.setMinEvictableIdleTimeMillis(60000);    // 空闲60秒后可驱逐
        
        // 配合心跳配置，确保连接池中的连接不会被防火墙断开
    }
}

要点总结

• 心跳是定期交换的控制帧，用于检测连接活性
• 客户端提议心跳间隔，服务端可拒绝，最终取较大值
• 连续2次未收到心跳帧即判定连接失效，触发重连
• 心跳间隔推荐30~60秒，过短增加CPU负担，过长延迟故障检测
• 启用自动恢复（setAutomaticRecoveryEnabled）实现断线重连
• 启用Topology恢复（setTopologyRecoveryEnabled）重建队列和交换器
• 心跳可防止防火墙/NAT断开空闲连接，起到保活作用
• 连接池需配合心跳使用，定期检查空闲连接有效性