如何让持续Azure WebJob接收队列新消息通知并实现高低优先级队列切换？

我来帮你解决这两个Azure WebJob的实际问题，都是日常开发里很常见的场景：

问题1：如何让持续运行的Azure WebJob接收到队列中新消息的通知？ #

你有两种靠谱的方案，选哪种取决于你是否使用Azure WebJobs SDK：

• 用WebJobs SDK的QueueTrigger（最省心）：SDK会自动帮你监听目标队列，一旦有新消息就触发对应的处理方法，完全不用自己写轮询逻辑。你只需要在处理方法上标记[QueueTrigger("your-queue-name")]，SDK底层用的是Azure Storage的长轮询机制——也就是保持连接等待消息，有消息就立即返回，超时后再重新发起请求，既保证了实时性，又不会产生不必要的IO计费。这种方式完美适配持续运行的WebJob场景。

• 手动实现长轮询（不依赖SDK）：如果不想用SDK，自己写持续运行的逻辑也可以。用Azure Storage Queue的ReceiveMessageAsync方法时，设置一个30秒左右的等待时间（比如TimeSpan.FromSeconds(30)），这就是长轮询。你的程序会一直保持和队列的连接，直到有消息进来或者超时，超时后再重试。这样能做到有消息就及时接收，比频繁短轮询省很多IO费用。示例代码大概是这样：

var queueClient = new QueueClient(yourStorageConnString, "your-queue");
while (true)
{
    var messageResponse = await queueClient.ReceiveMessageAsync(TimeSpan.FromSeconds(30));
    if (messageResponse.Value != null)
    {
        // 处理你的消息逻辑
        await queueClient.DeleteMessageAsync(messageResponse.Value.MessageId, messageResponse.Value.PopReceipt);
    }
}

问题2：优先级队列切换处理，避免频繁检查高优先级队列 #

你的核心诉求是不要每次处理普通消息前都查高优先级队列（毕竟每次查询都是计费IO），那可以用「双监听+同步控制」的思路来实现，具体步骤如下：

1. 拆分两个独立的监听任务：一个专门盯高优先级队列，另一个盯普通队列；
2. 用线程安全的标记控制流程：当高优先级队列有消息时，标记「正在处理高优先级」，让普通队列的监听暂停接收新消息；等高优先级队列的消息全部处理完，再清除标记，恢复普通队列的处理；
3. 高优先级队列用长轮询监听：只有当队列有消息时才会触发处理，不会产生额外的无效查询费用。

给你一个具体的代码示例，逻辑很清晰：

// 线程安全的标记，控制是否正在处理高优先级消息
private static bool _isProcessingHighPriority = false;
private static readonly object _syncLock = new object();
private static QueueClient _highPriorityQueue;
private static QueueClient _normalQueue;

public static async Task Main()
{
    var storageConnString = Environment.GetEnvironmentVariable("AzureWebJobsStorage");
    _highPriorityQueue = new QueueClient(storageConnString, "high-priority-queue");
    _normalQueue = new QueueClient(storageConnString, "normal-priority-queue");

    // 同时启动两个监听任务
    var highPriorityListener = MonitorHighPriorityQueue();
    var normalPriorityListener = MonitorNormalPriorityQueue();

    await Task.WhenAll(highPriorityListener, normalPriorityListener);
}

private static async Task MonitorHighPriorityQueue()
{
    while (true)
    {
        // 长轮询高优先级队列，有消息才返回
        var messageResponse = await _highPriorityQueue.ReceiveMessageAsync(TimeSpan.FromSeconds(30));
        if (messageResponse.Value != null)
        {
            // 标记开始处理高优先级
            lock (_syncLock)
            {
                _isProcessingHighPriority = true;
            }

            // 处理当前消息
            await ProcessHighPriorityMessage(messageResponse.Value);
            await _highPriorityQueue.DeleteMessageAsync(messageResponse.Value.MessageId, messageResponse.Value.PopReceipt);

            // 循环处理高优先级队列的剩余消息，直到为空
            while (true)
            {
                var nextMessage = await _highPriorityQueue.ReceiveMessageAsync();
                if (nextMessage.Value == null)
                {
                    // 高优先级队列为空，恢复普通队列处理
                    lock (_syncLock)
                    {
                        _isProcessingHighPriority = false;
                        Monitor.Pulse(_syncLock); // 通知普通队列可以继续了
                    }
                    break;
                }
                await ProcessHighPriorityMessage(nextMessage.Value);
                await _highPriorityQueue.DeleteMessageAsync(nextMessage.Value.MessageId, nextMessage.Value.PopReceipt);
            }
        }
    }
}

private static async Task MonitorNormalPriorityQueue()
{
    while (true)
    {
        // 检查是否正在处理高优先级，如果是就等待
        lock (_syncLock)
        {
            while (_isProcessingHighPriority)
            {
                Monitor.Wait(_syncLock); // 释放锁并等待通知
            }
        }

        // 长轮询普通队列
        var messageResponse = await _normalQueue.ReceiveMessageAsync(TimeSpan.FromSeconds(30));
        if (messageResponse.Value != null)
        {
            await ProcessNormalMessage(messageResponse.Value);
            await _normalQueue.DeleteMessageAsync(messageResponse.Value.MessageId, messageResponse.Value.PopReceipt);
        }
    }
}

// 高优先级消息处理逻辑
private static async Task ProcessHighPriorityMessage(QueueMessage message)
{
    Console.WriteLine($"Processing high-priority message: {message.Body}");
    await Task.Delay(1000); // 模拟处理耗时
}

// 普通优先级消息处理逻辑
private static async Task ProcessNormalMessage(QueueMessage message)
{
    Console.WriteLine($"Processing normal message: {message.Body}");
    await Task.Delay(1000); // 模拟处理耗时
}

这个方案的优势在于：

• 高优先级队列用长轮询，只有有消息时才会触发处理，完全避免了无效的查询计费；
• 用锁和等待/通知机制控制普通队列的暂停和恢复，不会在处理普通消息前反复查询高优先级队列；
• 逻辑清晰，完全符合你「处理完高优先级所有消息再回到普通队列」的需求。

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