Spring Boot RabbitMQ监听器Redisson锁无法解锁引发消息重复投递

问题根源 #

你的代码存在两个核心问题：

1. 未获取锁就执行解锁操作：代码仅创建了RLock实例，但从未调用lock()或tryLock()方法获取锁，直接在finally块执行lock.unlock()，导致当前线程未持有锁，触发IllegalMonitorStateException。
2. 异常引发消息无限重复投递：监听器抛出异常后，RabbitMQ默认将消息重新入队并重复投递，形成循环。

修正方案 #

1. 先获取锁再执行业务逻辑 #

必须调用RLock的锁获取方法，同时在解锁前判断当前线程是否持有锁，避免解锁未持有的锁。

2. 手动控制RabbitMQ消息确认 #

开启手动确认模式，根据业务结果决定消息的确认方式，避免无意义的重复投递。

修正后的代码示例 #

@RabbitListener(queues = Const.Rabbit.QUEUE_NAME, concurrency = "1")
public void consumer(byte[] payload, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
    String lockName = "lock-name";
    RLock lock = redissonClient.getLock(lockName);
    boolean isLocked = false;

    try {
        // 尝试获取锁：5秒内等待获取，锁持有30秒后自动释放
        isLocked = lock.tryLock(5, 30, TimeUnit.SECONDS);
        if (isLocked) {
            // 执行业务逻辑
            // call a method
            
            // 业务执行成功，手动确认消息
            channel.basicAck(deliveryTag, false);
        } else {
            // 获取锁失败，将消息重新入队，稍后重试
            channel.basicNack(deliveryTag, false, true);
            log.warn("无法获取锁，消息将重新入队");
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        log.error("获取锁时被中断", e);
        channel.basicNack(deliveryTag, false, true);
    } catch (Exception e) {
        log.error("业务执行出错: {}", e.getMessage());
        // 业务异常，拒绝消息并发送到死信队列（根据需求调整）
        channel.basicNack(deliveryTag, false, false);
    } finally {
        // 仅当当前线程持有锁时才解锁
        if (isLocked && lock.isHeldByCurrentThread()) {
            lock.unlock();
        }
    }
}

关键注意点 #

• 锁获取方式选择：如果需要阻塞等待锁，使用lock.lock()；如果不想阻塞，用tryLock()指定超时时间，避免线程长时间阻塞。
• 锁自动过期：Redisson锁支持自动过期，防止服务宕机后锁无法释放，建议设置合理的过期时间。
• 消息确认策略：根据业务场景选择：
  • basicAck：确认消息已处理，RabbitMQ会删除消息。
  • basicNack：拒绝消息，第三个参数requeue为true时消息重新入队，为false时消息会被丢弃或进入死信队列。

内容的提问来源于stack exchange，提问作者Reza Shahazegh