Azure Service Bus：已处理消息保留在订阅中30天且不重复读取

要实现保留已处理消息30天且订阅者不重复读取的需求，核心在于调整Azure Service Bus的订阅配置和Azure Function的消息处理逻辑，以下是两种可行方案：

方案一：使用归档队列（推荐，符合Service Bus最佳实践） #

这种方式将已处理消息转移到专门的归档存储，原订阅仅处理未处理的消息，逻辑清晰且避免冗余：

1. 创建归档队列：新建一个Service Bus队列，通过Azure门户/CLI/ARM模板设置其DefaultMessageTimeToLive为30天，确保消息保留30天后自动过期删除。
2. 配置消息转发：
   • 自动方式：在原订阅的配置中开启自动转发，目标设置为新建的归档队列。消息处理完成后会自动转发到归档队列，同时从原订阅中移除。
   • 手动方式：在Function代码处理完消息后，手动将消息发送到归档队列。
3. 保持Function默认配置：无需修改AutoComplete（默认值为true），函数处理成功后会自动完成消息，确保原订阅不会重复读取该消息。

方案二：在原订阅中保留并标记已处理 #

如果必须在原订阅中保留消息，可通过标记+延迟消息实现：

1. 设置订阅保留期：在Azure门户将原订阅的DefaultMessageTimeToLive设为30天。
2. 修改Function触发配置：在ServiceBusTrigger属性中添加AutoComplete = false，禁止自动完成消息。
3. 更新Function代码：将参数改为Message类型，处理时标记消息为已处理并延迟（延迟后的消息不会出现在活跃消息列表中，但仍保留在订阅内直到过期）：

public class MyFunction
{
   private readonly ILogger<MyFunction> _logger;

   public MyFunction(ILogger<MyFunction> logger)
   {
    _logger = logger;
   }

[Function(nameof(MyFunction))]
public async Task Run(
    [ServiceBusTrigger("%TopicName%", "%SubscriptionName%", 
        Connection = "ServiceBus.ConnectionString",
        AutoComplete = false)] 
    Message message,
    IMessageReceiver messageReceiver)
{
    try
    {
       _logger.LogInformation("Started !");
       
       // 检查消息是否已处理
       if (message.UserProperties.TryGetValue("IsProcessed", out var isProcessed) && (bool)isProcessed)
       {
           await messageReceiver.DeferAsync(message.SystemProperties.LockToken);
           return;
       }

       // 消息处理逻辑
       JObject jsonObject = JObject.Parse(Encoding.UTF8.GetString(message.Body));
       
       // 标记消息为已处理
       message.UserProperties["IsProcessed"] = true;
       // 延迟消息，避免重复接收，同时保留在订阅中
       await messageReceiver.DeferAsync(message.SystemProperties.LockToken);
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "消息处理失败");
        // 处理失败时将消息死信，避免反复重试
        await messageReceiver.DeadLetterAsync(message.SystemProperties.LockToken);
    }
  }
}

关键说明 #

• 你当前消息消失的原因是AutoComplete = true（默认配置）：Function处理成功后会自动调用CompleteAsync，将消息从订阅中移除。
• 延迟消息（DeferAsync）会将消息从活跃列表中移除，但仍保留在订阅内，直到达到DefaultMessageTimeToLive后自动删除；且只有通过ReceiveDeferredMessageAsync才能读取，确保普通订阅者不会重复接收。

内容的提问来源于stack exchange，提问作者the smart life