RabbitMQ实战应用

cheniy2025/8/15大约 3 分钟

RabbitMQ实战应用

前置知识

在学习本教程之前，请确保您已经：

掌握RabbitMQ的基础知识
了解RabbitMQ的高级特性
熟悉Linux系统操作
具备分布式系统基础知识

集群配置

1. 集群模式

RabbitMQ支持以下集群模式：

普通集群：队列数据只存在于单个节点
镜像集群：队列数据在多个节点之间同步
仲裁队列：基于Raft协议的新一代高可用方案

2. 普通集群配置

# 节点1配置
rabbitmq-server -detached

# 节点2配置
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# 节点3配置
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

3. 镜像集群策略

# 设置镜像策略
rabbitmqctl set_policy ha-all \
    "^" \
    '{"ha-mode":"all"}' \
    --priority 0 \
    --apply-to queues

高可用方案

1. HAProxy负载均衡

listen rabbitmq_cluster
    bind *:5672
    mode tcp
    balance roundrobin
    server rabbit1 192.168.1.101:5672 check inter 2000 rise 2 fall 3
    server rabbit2 192.168.1.102:5672 check inter 2000 rise 2 fall 3
    server rabbit3 192.168.1.103:5672 check inter 2000 rise 2 fall 3

2. Keepalived配置

vrrp_script chk_haproxy {
    script "killall -0 haproxy"
    interval 2
    weight 2
}

vrrp_instance VI_1 {
    state MASTER
    interface eth0
    virtual_router_id 51
    priority 100
    authentication {
        auth_type PASS
        auth_pass 1111
    }
    track_script {
        chk_haproxy
    }
    virtual_ipaddress {
        192.168.1.100
    }
}

3. 应用配置

spring:
  rabbitmq:
    addresses: 192.168.1.101:5672,192.168.1.102:5672,192.168.1.103:5672
    username: guest
    password: guest
    virtual-host: /
    connection-timeout: 15000
    # 开启重试
    template:
      retry:
        enabled: true
        initial-interval: 1000
        max-attempts: 3
        max-interval: 10000
        multiplier: 1.0

性能调优

1. 生产者优化

生产者配置示例

@Configuration
public class RabbitProducerConfig {
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        
        // 开启发送者确认
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.error("消息发送失败：{}", cause);
                // 重试发送逻辑
            }
        });
        
        // 开启发送者返回
        template.setReturnsCallback(returned -> {
            log.error("消息路由失败：{}", returned.getMessage());
            // 处理无法路由的消息
        });
        
        // 设置消息转换器
        template.setMessageConverter(new Jackson2JsonMessageConverter());
        
        return template;
    }
}


### 2. 消费者优化

::: details 消费者配置示例
```java
@Configuration
public class RabbitConsumerConfig {
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        
        // 设置并发消费者数量
        factory.setConcurrentConsumers(5);
        factory.setMaxConcurrentConsumers(10);
        
        // 设置预取数量
        factory.setPrefetchCount(1);
        
        // 设置消息转换器
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        
        // 设置手动确认
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        
        return factory;
    }
}

:::


### 3. 连接池优化

::: details 连接池配置示例
```java
@Configuration
public class RabbitConnectionConfig {
    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setHost("localhost");
        
        // 设置连接池大小
        factory.setChannelCacheSize(25);
        
        // 设置连接超时
        factory.setConnectionTimeout(30000);
        
        // 设置心跳间隔
        factory.setRequestedHeartbeat(Duration.ofSeconds(60));
        
        return factory;
    }
}

:::


## 监控告警

### 1. 监控指标

::: tip 关键指标
1. 队列深度
2. 消息处理延迟
3. 消费者数量
4. 连接数和信道数
5. 磁盘空间使用
:::

### 2. Prometheus集成

```yaml
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'rabbitmq'
    static_configs:
      - targets: ['localhost:15692']

3. 告警规则

groups:
- name: rabbitmq_alerts
  rules:
  - alert: QueueDepthTooHigh
    expr: rabbitmq_queue_messages > 10000
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: Queue depth too high
      description: Queue {{ $labels.queue }} has more than 10k messages