Rebus是否支持查询队列中处于延迟状态的指定类型消息？

Great question—this is a super common scenario with multi-instance services and scheduled high-cost jobs, so let’s break down your options clearly.

Option 1: Avoid the Need to Query (Highly Recommended) #

Instead of checking for existing delayed messages (which can introduce tricky race conditions), you can leverage message uniqueness and Azure Service Bus’s duplicate detection to guarantee only one instance of your daily task message exists. Here’s how to pull it off:

• When sending your delayed message, assign a unique MessageId tied to the task type and execution date (e.g., DailyHighCostOperation-2024-05-20). This makes each day’s task have a distinct, predictable ID.
• Enable duplicate detection on your Azure Service Bus queue with a window larger than 24 hours (since your task runs daily). Azure Service Bus will automatically discard any subsequent messages with the same MessageId within this window.

With Rebus, you can set the MessageId directly when sending:

await bus.Advanced.SendWithRouting(
    destinationQueueName,
    new DailyHighCostTask(),
    options =>
    {
        options.SetMessageId($"DailyHighCostOperation-{DateTime.UtcNow:yyyy-MM-dd}");
        options.DeferUntil(DateTime.UtcNow.AddHours(2)); // Your desired scheduled time
    });

This approach is way more reliable than querying because it eliminates the race condition where two instances might check for the message at the exact same time, both decide it doesn’t exist, and both send it.

Option 2: Query for Existing Delayed Messages (If You Have To) #

Rebus is built as an abstraction over message transports, so it doesn’t expose a built-in API to query for delayed messages (since different transports handle this logic differently). If you absolutely need to check for existing delayed messages of your specific type, you’ll need to use the Azure Service Bus Administration SDK directly.

Here’s a quick example of how to peek at delayed messages in your queue and filter by message type (assuming you’ve added a custom property to identify the message type):

using Azure.Messaging.ServiceBus.Administration;

var adminClient = new ServiceBusAdministrationClient("your-connection-string");
var peekResult = await adminClient.PeekMessagesAsync(
    queueName: "your-queue-name",
    maxMessages: 50, // Adjust based on your expected message volume
    fromEnqueuedTime: DateTimeOffset.UtcNow.AddDays(-1) // Look back 1 day to cover the daily window
);

var hasExistingDelayedTask = peekResult.Any(message =>
    // Check if the message is still scheduled for the future (delayed)
    message.ScheduledEnqueueTime > DateTimeOffset.UtcNow
    // Check your custom message type property to target the right task
    && message.ApplicationProperties.TryGetValue("MessageType", out var type)
    && type.ToString() == typeof(DailyHighCostTask).FullName
);

Just a heads-up: peeking at messages doesn’t remove them from the queue, so this is totally safe for checking purposes.

Final Takeaway #

Stick with Option 1 whenever you can—it aligns with message queue best practices, avoids race conditions, and keeps your code clean. Only use the direct Azure Service Bus query if you have a specific constraint that makes duplicate detection unworkable.

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