202528 | RabbitMQ-高级 | 消息可靠性 | 业务幂等性 | 延迟消息

消息可靠性

RabbitMQ发送者可靠性

一、发送者重连机制

1. 核心配置（application.yml）

spring:
  rabbitmq:
    addresses: rabbit1:5672,rabbit2:5672 # 集群地址
    connection-timeout: 5000    # 连接超时(ms)
    template:
      retry:
        enabled: true           # 启用重试
        max-attempts: 3         # 最大重试次数
        initial-interval: 1000  # 初始间隔(ms)
        multiplier: 2           # 间隔乘数

2. 高级重连配置（Java Config）

@Bean
public CachingConnectionFactory connectionFactory() {
    CachingConnectionFactory factory = new CachingConnectionFactory();
    factory.setHost("cluster.rabbitmq.cn");
    
    // 重连关键参数
    factory.setAutomaticRecoveryEnabled(true);  // 自动恢复
    factory.setNetworkRecoveryInterval(5000);   // 5秒重试间隔
    factory.setRequestedHeartbeat(30);          // 30秒心跳
    
    // 异常处理器
    factory.setRecoveryListener(new RecoveryListener() {
        @Override
        public void handleRecovery(Recoverable recoverable) {
            log.info("连接已恢复");
        }
        
        @Override
        public void handleRecoveryStarted(Recoverable recoverable) {
            log.warn("开始重连...");
        }
    });
    
    return factory;
}

3. 重连过程示意图

二、发送者确认机制

1. 确认模式配置

@Configuration
public class RabbitConfirmConfig {
    
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        
        // 1. 启用Confirm模式
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        
        // 2. 设置Confirm回调
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                metricService.increment("mq.ack.success");
            } else {
                log.error("消息未确认 ID: {}, 原因: {}", 
                    correlationData.getId(), cause);
                retryQueue.add(correlationData);
            }
        });
        
        // 3. 启用Return回调
        template.setMandatory(true);
        template.setReturnsCallback(returned -> {
            log.warn("消息不可路由: {}", returned.getMessage());
            deadLetterService.save(returned);
        });
        
        return template;
    }
}

2. 消息发送示例

@Service
public class OrderSenderService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendOrder(Order order) {
        // 1. 构建唯一ID
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        
        // 2. 发送消息（携带correlationData）
        rabbitTemplate.convertAndSend(
            "order.exchange", 
            "order.create", 
            order,
            message -> {
                // 添加自定义头
                message.getMessageProperties()
                       .setHeader("retry-count", 0);
                return message;
            },
            correlationData);
    }
}

3. 确认流程示意图

三、生产级完整方案

1. 消息状态追踪设计

public class MessageTracker {
    private static final ConcurrentMap<String, MessageRecord> records = 
        new ConcurrentHashMap<>();
    
    public static void track(String messageId, Message message) {
        records.put(messageId, new MessageRecord(message, System.currentTimeMillis()));
    }
    
    public static void confirm(String messageId) {
        records.get(messageId).confirm();
    }
    
    @Scheduled(fixedRate = 60000)
    public void checkTimeoutMessages() {
        records.values().stream()
            .filter(r -> !r.isConfirmed() && r.isTimeout())
            .forEach(this::resend);
    }
}

2. 混合可靠性配置

spring:
  rabbitmq:
    # 连接配置
    addresses: rabbit1:5672,rabbit2:5672
    connection-timeout: 10000
    
    # 发送者确认
    publisher-confirm-type: correlated
    publisher-returns: true
    
    # 模板配置
    template:
      mandatory: true
      retry:
        enabled: true
        max-attempts: 3

3. 异常处理流程图

RabbitMQ 中消息队列可靠性：

一、数据持久化（Message Durability）

1. 核心概念

数据持久化是 RabbitMQ 防止消息丢失的基础机制，通过将消息和元数据写入磁盘，确保 Broker 重启后数据不丢失。

2. 持久化三要素

3. Spring Boot 配置示例

// 持久化交换机
@Bean
public DirectExchange durableExchange() {
    return new DirectExchange("order.direct", true, false);
}

// 持久化队列
@Bean
public Queue durableQueue() {
    return new Queue("order.queue", true, false, false);
}

// 发送持久化消息
rabbitTemplate.convertAndSend("exchange", "key", message, msg -> {
    msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
    return msg;
});

4. 持久化性能影响

二、Lazy Queue（惰性队列）

1. 设计目的

解决内存溢出风险，通过将消息直接写入磁盘而非内存，适用于：

• 高吞吐但低优先级的消息（如日志）
• 可能产生消息堆积的场景

2. 核心特性

3. 配置方式

声明时指定（推荐）

@Bean
public Queue lazyQueue() {
    Map<String, Object> args = new HashMap();
    args.put("x-queue-mode", "lazy"); // 关键参数
    return new Queue("lazy.queue", true, false, false, args);
}

策略批量设置

# 将所有队列设为lazy模式
rabbitmqctl set_policy Lazy "^lazy\." '{"queue-mode":"lazy"}' --apply-to queues

4. 性能对比测试

三、生产环境最佳实践

1. 混合使用场景

2. 监控指标建议

3. 容灾方案设计

// 高可用组合方案
@Bean
public Queue highReliabilityQueue() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-queue-mode", "lazy"); // 防内存溢出
    args.put("x-message-ttl", 86400000); // 24小时TTL
    args.put("x-dead-letter-exchange", "dlx"); // 死信处理
    return new Queue("order.backup", true, false, false, args); // 持久化
}

四、常见问题解决方案

1. 持久化消息性能低下

• 优化方案：

  spring:
    rabbitmq:
      template:
        channel-cache-size: 32  # 增加通道缓存
      listener:
        direct:
          prefetch: 100         # 提高预取值
  

2. Lazy Queue消费延迟

• 优化方案：

  // 增加消费者并发
  @RabbitListener(queues = "lazy.queue", concurrency = "5")
  public void handleLazyMessage(Message msg) {
      // 处理逻辑
  }
  

3. 磁盘空间不足

• 应急处理：

  # 临时切换存储路径
  rabbitmqctl set_disk_free_limit 5GB
  rabbitmqctl stop_app
  rabbitmqctl reset
  rabbitmqctl start_app
  

通过合理结合 数据持久化 和 Lazy Queue，可以实现：

• 关键业务消息零丢失（持久化保障）
• 突发流量下的系统稳定性（Lazy Queue防内存溢出）
• 资源使用的智能平衡（根据业务特性混合配置）

RabbitMQ消费者可靠性全面指南

一、消费者确认机制（ACK/NACK）

1. 确认模式工作流程

2. ACK模式类型对比

3. 手动ACK最佳实践代码

@RabbitListener(queues = "orders")
public void handleOrder(Order order, 
                      Channel channel,
                      @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
    
    try {
        // 业务处理
        orderService.process(order);
        // 成功确认
        channel.basicAck(tag, false); 
    } catch (BusinessException e) {
        // 业务异常：重试
        channel.basicNack(tag, false, true);
    } catch (Exception e) {
        // 系统异常：死信队列
        channel.basicNack(tag, false, false);
    }
}

二、失败重试与恢复机制

1. 重试策略流程图解

. 消息恢复器实现方案

@Bean
public MessageRecoverer customRecoverer() {
    // 方案1：转发到死信交换机
    return new RepublishMessageRecoverer(rabbitTemplate, "dlx.exchange", "error");
    
    // 方案2：自定义处理
    return (message, cause) -> {
        errorRepository.save(new ErrorLog(message, cause));
        alertService.notifyAdmin(cause);
    };
}

三、生产环境配置方案

1. 完整配置模板

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 50
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1000ms
          multiplier: 2
          max-interval: 10s
          stateless: false
          message-recoverer: customRecoverer

2. 异常处理决策树

四、高级场景解决方案

1. 顺序消息保障方案

@RabbitListener(queues = "sequential.queue")
public void handleSequential(Order order, Channel channel, 
                           @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    
    // 获取分布式锁
    Lock lock = redissonClient.getLock("order:"+order.getId());
    
    try {
        lock.lock();
        orderService.process(order);
        channel.basicAck(tag, false);
    } finally {
        lock.unlock();
    }
}

2. 消息积压应急方案

五、监控与告警

关键监控指标

RabbitMQ业务幂等性实现方案详解

一、幂等性核心概念

在消息队列系统中，业务幂等性是指：

1. 重复消费同一条消息不会导致业务数据错误
2. 多次处理相同请求与单次处理效果一致
3. 系统状态变更只发生一次

二、实现方案对比

三、具体实现方案

1. 消息ID去重方案

生产者配置：

// 发送消息时添加唯一ID
MessageProperties props = new MessageProperties();
props.setHeader("msg_id", UUID.randomUUID().toString());
Message message = new Message(body, props);
rabbitTemplate.send(exchange, routingKey, message);

消费者实现：

@RabbitListener(queues = "order.queue")
public void handleOrder(Message message) {
    String msgId = message.getMessageProperties().getHeader("msg_id");
    
    // Redis原子操作实现
    Boolean isNew = redisTemplate.opsForValue()
        .setIfAbsent("msg:" + msgId, "1", 24, TimeUnit.HOURS);
    
    if(Boolean.FALSE.equals(isNew)) {
        log.warn("重复消息已忽略: {}", msgId);
        return;
    }
    
    orderService.process(message.getBody());
}

2. 业务状态检查方案

public void processPayment(PaymentMessage message) {
    PaymentRecord record = paymentDao.findByOrderId(message.getOrderId());
    
    // 状态检查
    if(record != null && record.getStatus() == PaymentStatus.SUCCESS) {
        log.info("订单{}已支付，跳过处理", message.getOrderId());
        return;
    }
    
    // 执行业务逻辑
    boolean result = paymentGateway.charge(message);
    paymentDao.save(new PaymentRecord(message, result));
}

3. 数据库幂等方案

建表SQL：

CREATE TABLE transactions (
    id VARCHAR(64) PRIMARY KEY,
    order_id BIGINT UNIQUE,
    status VARCHAR(20),
    created_at TIMESTAMP
);

JPA实现：

@Transactional
public void processTransaction(TransactionMessage message) {
    // 先检查后插入
    if(transactionRepository.existsById(message.getTxId())) {
        return;
    }
    
    // 使用数据库唯一约束
    try {
        transactionRepository.save(
            new Transaction(
                message.getTxId(),
                message.getOrderId(),
                "PROCESSING"
            )
        );
        // 业务处理...
    } catch (DataIntegrityViolationException e) {
        log.warn("重复交易: {}", message.getTxId());
    }
}

四、生产环境最佳实践

1. 复合幂等策略

public void handleOrder(OrderMessage message) {
    // 第一层：消息ID检查
    if(idempotentService.isMessageProcessed(message.getId())) {
        return;
    }
    
    // 第二层：业务状态检查
    Order order = orderService.getOrder(message.getOrderId());
    if(order.isPaid()) {
        return;
    }
    
    // 第三层：数据库乐观锁
    try {
        orderService.processWithLock(order);
    } catch (OptimisticLockingFailureException e) {
        log.error("并发处理订单: {}", order.getId());
    }
}

2. 异常处理方案

3. 监控指标设计

五、常见问题解决方案

1. Redis宕机时的降级方案

public boolean checkMessageId(String msgId) {
    try {
        // 优先使用Redis
        return redisTemplate.opsForValue().setIfAbsent(msgId, "1", 24, HOURS);
    } catch (Exception e) {
        // 降级到数据库检查
        return !messageLogRepository.existsById(msgId);
    }
}

2. 分布式锁实现

public void processWithLock(String orderId) {
    Lock lock = redissonClient.getLock("order:" + orderId);
    try {
        if(lock.tryLock(3, 30, TimeUnit.SECONDS)) {
            // 临界区代码
            orderService.process(orderId);
        }
    } finally {
        lock.unlock();
    }
}

3. 消息追溯方案

@Aspect
public class MessageTraceAspect {
    @AfterReturning("execution(* com..*Listener.*(..)) && args(message)")
    public void afterMessage(Message message) {
        auditService.record(
            message.getMessageProperties().getHeader("msg_id"),
            "PROCESSED",
            LocalDateTime.now()
        );
    }
}

RabbitMQ延时消息实现方案

一、核心实现方案对比

二、死信队列方案实现（原生支持）

1. 架构设计

2. 具体实现代码

配置声明：

@Configuration
public class DelayQueueConfig {
    // 死信交换机
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange("dlx.exchange");
    }

    // 死信队列
    @Bean
    public Queue delayQueue() {
        return new Queue("delay.queue");
    }

    // 业务队列（设置死信参数）
    @Bean
    public Queue businessQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", "dlx.exchange");
        args.put("x-dead-letter-routing-key", "delay.key");
        return new Queue("order.queue", true, false, false, args);
    }

    // 绑定关系
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(delayQueue())
               .to(dlxExchange())
               .with("delay.key");
    }
}

发送延时消息：

public void sendDelayMessage(Order order, int delayMinutes) {
    // 设置消息属性
    MessageProperties props = new MessageProperties();
    props.setExpiration(String.valueOf(delayMinutes * 60 * 1000)); // 毫秒
    
    // 发送消息
    rabbitTemplate.convertAndSend(
        "order.queue", 
        new Message(order.toString().getBytes(), props)
    );
}

三、RabbitMQ插件方案（推荐方案）

1. 安装延时插件

# 下载插件（版本需匹配）
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.11.1/rabbitmq_delayed_message_exchange-3.11.1.ez

# 启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

2. 具体实现代码

配置延时交换机：

@Bean
public CustomExchange delayExchange() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-delayed-type", "direct");
    return new CustomExchange(
        "delayed.exchange", 
        "x-delayed-message", 
        true, 
        false, 
        args
    );
}

发送延时消息：

public void sendDelayMessage(Order order, int delaySeconds) {
    rabbitTemplate.convertAndSend(
        "delayed.exchange",
        "order.routing.key",
        order,
        message -> {
            message.getMessageProperties()
                  .setHeader("x-delay", delaySeconds * 1000);
            return message;
        }
    );
}

四、完整订单超时关闭案例

1. 业务场景

• 订单创建后30分钟未支付自动关闭
• 支付成功后取消延时任务

2. 实现方案设计

3. 关键代码实现

订单服务生产者：

public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void createOrder(Order order) {
        // 保存订单
        orderRepository.save(order);
        
        // 发送延时消息（插件方案）
        rabbitTemplate.convertAndSend(
            "delayed.exchange",
            "order.close",
            order.getId(),
            message -> {
                message.getMessageProperties()
                      .setHeader("x-delay", 30 * 60 * 1000); // 30分钟
                return message;
            }
        );
    }

    public void cancelTimeoutTask(String orderId) {
        // 支付成功后删除消息（需要消息ID）
        // 实际实现需要改造发送逻辑保存messageId
        rabbitTemplate.execute(channel -> {
            channel.queuePurge("order.close.queue");
            return null;
        });
    }
}

消费者实现：

@RabbitListener(queues = "order.close.queue")
public void handleTimeoutOrder(String orderId) {
    Order order = orderRepository.findById(orderId);
    if(order.getStatus() == OrderStatus.UNPAID) {
        orderService.closeOrder(orderId);
        log.info("订单超时关闭: {}", orderId);
    }
}