Azure Service Bus的IMessageReceiver.receiveBatch()重复读取DLQ同批次消息问题

问题根源 #

• 默认接收模式为PeekLock：Azure Service Bus 消息接收默认使用PeekLock模式，调用receiveBatch获取到的消息仅被临时锁定，不会直接从队列中移除。如果锁定超时前没有主动调用complete方法确认消费，消息会自动重新回到队列，导致下一批读取时重复拿到相同消息。
• 批量处理逻辑阻塞：代码中executor.invokeAll会阻塞当前线程，直到该批次所有消息的处理任务全部执行完成，因此会出现第一批未处理完无法读取下一批的现象。

解决方案 #

1. 消费成功后主动确认消息 #

在handleDeadLetterMessage方法中，单条消息处理完成后调用complete方法，通知服务端该消息已成功消费，可以从队列中移除：

private void handleDeadLetterMessage(IMessage message, IMessageReceiver deadLetterReceiver) {
    try {
        // 原有处理DLQ消息的业务逻辑
        doYourBusiness(message);
        
        // 处理成功，确认消费，消息会从DLQ中移除
        deadLetterReceiver.complete(message.getLockToken());
    } catch (Exception e) {
        log.error("处理DLQ消息失败，messageId:{}", message.getMessageId(), e);
        // 处理失败的可选操作：
        // 1. 放回队列重试：deadLetterReceiver.abandon(message.getLockToken())
        // 2. 直接丢弃无效消息：deadLetterReceiver.complete(message.getLockToken())
        // 3. 超过重试次数后丢弃：自行维护重试次数判断
    }
}

2. 调整批量读取阻塞逻辑（可选，需非阻塞读取时使用） #

如果需要第一批未处理完就可以读取下一批，不要调用invokeAll后同步等待所有future.get返回，可以调整提交逻辑，同时通过线程池大小控制并发上限即可：

public void processDeadLetterQueue(){
    IMessageReceiver deadLetterReceiver = getDeadLetterMessageReceiver();
    Long deadLetterMessageCount = getDeadLetterMessageCount();
    Long receivedMessageCount = 0L;
    // 线程池大小根据业务处理能力调整
    ExecutorService executor = Executors.newFixedThreadPool(10);
 
    while(receivedMessageCount < deadLetterMessageCount) {
        Collection<IMessage> messageList = deadLetterReceiver.receiveBatch(5);
        receivedMessageCount += messageList.size();
        messageList.forEach(message -> executor.submit(() -> {
            handleDeadLetterMessage(message, deadLetterReceiver);
            return null;
        }));
    }
    // 等待所有任务处理完成再关闭资源
    executor.shutdown();
    try {
        executor.awaitTermination(1, TimeUnit.HOURS);
    } catch (InterruptedException e) {
        log.error("等待任务处理完成中断", e);
        Thread.currentThread().interrupt();
    }
    deadLetterReceiver.close();
}

注意事项 #

• 锁超时续约：如果单条消息处理时间超过队列设置的锁有效期（默认30秒，最大可设为5分钟），需要在处理过程中调用deadLetterReceiver.renewLock(message.getLockToken())主动续约，避免锁提前过期导致消息回流。
• 消息幂等：极端情况下如果网络异常导致complete请求没有成功发到服务端，消息仍可能被重复投递，业务逻辑需要做好幂等处理。

内容的提问来源于stack exchange，提问作者Vinayaka S P