Managed Queues and Messaging
Read work from SQS or Service Bus in a BackgroundService: long polling, visibility timeouts and dead-letter queues.
await sqs.ReceiveMessageAsync(new ReceiveMessageRequest { QueueUrl = url, WaitTimeSeconds = 20 }, ct);A queue is a service that holds messages. One process sends a message; another receives it and does the work later. That lets a web request finish straight away, and lets the worker side scale on its own. Amazon SQS and Azure Service Bus are both managed queues, which means the provider runs them for you.
Receive, work, then delete
Receiving a message does not remove it. It hides it for a period called the visibility timeout, so no other worker sees it. Deleting the message is what records the work as done. If your process dies first, the timeout expires and the message comes back. That is at-least-once delivery: every message arrives, some arrive twice, so a handler must be safe to run twice.
Long polling and dead letters
Long polling means asking the queue to hold your request open until a message turns up, up to 20 seconds on SQS through WaitTimeSeconds, instead of answering "nothing here" at once. Fewer calls, lower latency, lower bill. A dead-letter queue is a second queue the service moves a message to after it has been received a set number of times without being deleted. Without one, a message that can never succeed is redelivered forever.
What trips people up
Work that outlasts the visibility timeout is handed to a second worker while the first is still running. Either set the timeout above your slowest job, or extend it as you go.
Example
// dotnet add package AWSSDK.SQS
// dotnet add package AWSSDK.Extensions.NETCore.Setup
// dotnet add package Microsoft.Extensions.Hosting
using Amazon.Extensions.NETCore.Setup;
using Amazon.SQS;
using Amazon.SQS.Model;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
HostApplicationBuilder builder = Host.CreateApplicationBuilder(args);
builder.Services.AddDefaultAWSOptions(builder.Configuration.GetAWSOptions());
builder.Services.AddAWSService<IAmazonSQS>();
builder.Services.AddHostedService<OrderWorker>();
await builder.Build().RunAsync();
public sealed class OrderWorker(IAmazonSQS sqs, ILogger<OrderWorker> logger) : BackgroundService
{
private const string QueueUrl =
"https://sqs.eu-west-1.amazonaws.com/111122223333/orders";
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
// WaitTimeSeconds 20 is long polling. VisibilityTimeout 60 is how
// long this worker has to finish before the message returns.
ReceiveMessageResponse batch = await sqs.ReceiveMessageAsync(
new ReceiveMessageRequest
{
QueueUrl = QueueUrl,
MaxNumberOfMessages = 10,
WaitTimeSeconds = 20,
VisibilityTimeout = 60
},
stoppingToken);
// In version 4 of the AWS SDK an empty batch leaves Messages null.
foreach (Message message in batch.Messages ?? [])
{
logger.LogInformation("handling {MessageId}", message.MessageId);
await sqs.DeleteMessageAsync(
new DeleteMessageRequest
{
QueueUrl = QueueUrl,
ReceiptHandle = message.ReceiptHandle
},
stoppingToken);
}
}
}
}When to use it
- Send the order confirmation email after checkout without making the customer wait for the mail server.
- Spread 50,000 PDF renders across however many worker pods happen to be running.
- Hold work safely while a downstream service is down, instead of dropping it.
More examples
Azure Service Bus, and dead-lettering on purpose
Service Bus gives you the receive loop. CompleteMessageAsync is the equivalent of the SQS delete, and setting AutoCompleteMessages to false keeps that decision yours. Dead-letter a message you know can never succeed rather than letting it burn through its delivery attempts.
// dotnet add package Azure.Messaging.ServiceBus
using Azure.Identity;
using Azure.Messaging.ServiceBus;
await using ServiceBusClient client = new(
"contoso.servicebus.windows.net", new DefaultAzureCredential());
await using ServiceBusProcessor processor = client.CreateProcessor("orders",
new ServiceBusProcessorOptions
{
MaxConcurrentCalls = 4,
AutoCompleteMessages = false
});
processor.ProcessMessageAsync += async args =>
{
string body = args.Message.Body.ToString();
if (body.Length == 0)
{
// A retry will never help, so move it aside now.
await args.DeadLetterMessageAsync(args.Message, "empty body",
cancellationToken: args.CancellationToken);
return;
}
await args.CompleteMessageAsync(args.Message, args.CancellationToken);
};
// Service Bus requires an error handler; without one the processor will not start.
processor.ProcessErrorAsync += args =>
{
Console.WriteLine($"error: {args.Exception.Message}");
return Task.CompletedTask;
};
await processor.StartProcessingAsync();The sending side, with a key the handler can deduplicate on
The endpoint returns 202 Accepted as soon as the message is queued, which is the point of the queue. Carrying a stable key on every message is what lets the worker recognise a redelivery instead of charging a card twice.
using System.Text.Json;
using Amazon.SQS;
using Amazon.SQS.Model;
app.MapPost("/orders", async (Order order, IAmazonSQS sqs, CancellationToken ct) =>
{
await sqs.SendMessageAsync(new SendMessageRequest
{
QueueUrl = "https://sqs.eu-west-1.amazonaws.com/111122223333/orders",
MessageBody = JsonSerializer.Serialize(order),
MessageAttributes = new Dictionary<string, MessageAttributeValue>
{
// Attributes travel beside the body, so a worker can check this
// without parsing the JSON.
["idempotency-key"] = new MessageAttributeValue
{
DataType = "String",
StringValue = order.Reference
}
}
}, ct);
return Results.Accepted($"/orders/{order.Reference}");
});Buying more time for a slow message
Use this for jobs whose length you cannot predict, such as rendering a document of unknown size. The alternative, a very large visibility timeout on the queue itself, means a crashed worker's message sits invisible for that whole period before anyone retries it.
using Amazon.SQS;
using Amazon.SQS.Model;
// Call this partway through a job that may outrun its visibility timeout.
// The clock restarts from now, it does not add to what is left.
await sqs.ChangeMessageVisibilityAsync(new ChangeMessageVisibilityRequest
{
QueueUrl = queueUrl,
ReceiptHandle = receiptHandle,
VisibilityTimeout = 300
}, ct);
Discussion