Azure 服务总线未检测到重复项

Question

我有一个从 Azure 服务总线队列读取消息并将该消息转换为要由 Azure 媒体服务编码的视频的进程。我注意到，如果该过程连续快速启动，则同一视频将紧接着被编码。这是我将视频添加到队列的代码

public class VideoManager
{
    string _connectionString = ConfigurationManager.AppSettings["Microsoft.ServiceBus.ConnectionString"];
    string _queueName = ConfigurationManager.AppSettings["ServiceBusQueueName"];
    QueueClient _client;

    public VideoManager()
    {
        var conStringBuilder = new ServiceBusConnectionStringBuilder(_connectionString)
        {
            OperationTimeout = TimeSpan.FromMinutes(120)
        };

        var messagingFactory = MessagingFactory.CreateFromConnectionString(conStringBuilder.ToString());
        _client = messagingFactory.CreateQueueClient(_queueName);
    }

    public void Approve(Video video)
    {
        // Set video to approved. 
        video.ApprovalStatus = ApprovalStatus.Approved;
        var message = new BrokeredMessage(new VideoMessage(video, VideoMessage.MessageTypes.Approve, string.Empty));
        message.MessageId = video.RowKey;
        _client.Send(message);
    }
}

以及从队列中读取的进程

 class Program
{
    static QueueClient client;

    static void Main(string[] args)
    {
        VideoManager videoManager = new VideoManager();

        var connectionString = ConfigurationManager.AppSettings["Microsoft.ServiceBus.ConnectionString"];

        var conStringBuilder = new ServiceBusConnectionStringBuilder(connectionString)
        {
            OperationTimeout = TimeSpan.FromMinutes(120)
        };

        var messagingFactory = MessagingFactory.CreateFromConnectionString(conStringBuilder.ToString());

        client = messagingFactory.CreateQueueClient(ConfigurationManager.AppSettings["ServiceBusQueueName"]);

        Console.WriteLine("Starting: Broadcast Center Continuous Video Processing Job");

        OnMessageOptions options = new OnMessageOptions
        {
            MaxConcurrentCalls = 25,
            AutoComplete = false
        };

        client.OnMessageAsync(async message =>
        {
            bool shouldAbandon = false;

            try
            {
                await HandleMessage(message);
            }
            catch (Exception ex)
            {
                shouldAbandon = true;
                Console.WriteLine(ex.Message);
            }
            if (shouldAbandon)
            {
                await message.AbandonAsync();
            }
        }, options);
        while (true) { }
    }
    async static Task<int> HandleMessage(BrokeredMessage message)
    {

        VideoMessage videoMessage = message.GetBody<VideoMessage>();

        Console.WriteLine(String.Format("Message body: {0}", videoMessage.Video.Title));
        Console.WriteLine(String.Format("Message id: {0}", message.MessageId));

        VideoProcessingService vp = new VideoProcessingService(videoMessage.Video);
        Task task;
        switch (videoMessage.MessageType)
        {
            case VideoMessage.MessageTypes.CreateThumbnail:
                task = new Task(() => vp.ProcessThumbnail(videoMessage.TimeStamp));
                task.Start();

                while (!task.IsCompleted)
                {
                    await Task.Delay(15000);
                    message.RenewLock();
                }
                await task;
                Console.WriteLine(task.Status.ToString());

                Console.WriteLine("Processing Complete");
                Console.WriteLine("Awaiting Message");
                break;
            case VideoMessage.MessageTypes.Approve:

                task = new Task(() => vp.Approve());
                task.Start();

                while (!task.IsCompleted)
                {
                    await Task.Delay(15000);
                    message.RenewLock();
                }
                await task;
                Console.WriteLine(task.Status.ToString());

                Console.WriteLine("Processing Complete");
                Console.WriteLine("Awaiting Message");
                break;
            default:
                break;
        }
        return 0;
    }
}

如果我连续 3 次启动该进程，我在控制台窗口中看到的内容如下

消息 ID：76aca19a-0698-449b-bf58-a24876fc4314

消息 ID：76aca19a-0698-449b-bf58-a24876fc4314

消息 ID：76aca19a-0698-449b-bf58-a24876fc4314

我想也许我的设置不正确，但它们在那里 我在这里真的很茫然，因为我希望这是开箱即用的行为。重复检测是否只有在消息完成后才起作用，所以我不能使用 OnMessageAsync()？

Answer 1

问题不在于完成（就像在代码中那样），而是事实上您有多个消费者（25 个并发回调），而且看起来 LockDuration 的消逝速度比处理速度快。结果，消息重新出现并重新处理。因此，您会看到多次记录相同的消息 ID。

可能的解决方案是（正如我在上面的 comment 中概述的那样）：

1. 让 OnMessage API 为您管理超时扩展 (example)
2. 使用BrokeredMessage.RenewLock 手动更新锁

Answer 2

您的 HandleMessage 代码中缺少一行代码。

async static Task<int> HandleMessage(BrokeredMessage message)
{
  VideoMessage videoMessage = message.GetBody<VideoMessage>();

  message.CompleteAsync(); // This line...

  Console.WriteLine(String.Format("Message id: {0}", message.MessageId));
  // Processes Message
}

所以是的，您必须使用 Complete、Defer 等来标记消息。

另见Answer，还发现this，这可能对重复检测的工作原理有用