消息与后台任务

public class EmailQueue
{
    private readonly Channel<EmailMessage> _channel =
        Channel.CreateUnbounded<EmailMessage>();

    public void Enqueue(EmailMessage msg) => _channel.Writer.TryWrite(msg);
    public IAsyncEnumerable<EmailMessage> ReadAllAsync(CancellationToken ct) =>
        _channel.Reader.ReadAllAsync(ct);
}

public class EmailSenderWorker(EmailQueue queue, IEmailClient client)
    : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (var msg in queue.ReadAllAsync(stoppingToken))
        {
            try
            {
                await client.SendAsync(msg);
            }
            catch (Exception ex)
            {
                Console.WriteLine($"发送失败: {msg.To} - {ex.Message}");
            }
        }
    }
}

// 请求侧：投递后立即返回 202
app.MapPost("/contact", (EmailQueue queue, ContactForm form) =>
{
    queue.Enqueue(new EmailMessage(form.Email, "新咨询"));
    return Results.Accepted();
});

public class OrderDispatcher : BackgroundService
{
    private readonly Channel<OrderCreatedEvent> _channel;
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly int _concurrency;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var workers = Enumerable.Range(0, _concurrency)
            .Select(_ => ProcessAsync(stoppingToken));
        await Task.WhenAll(workers);   // 并发消费者
    }

    private async Task ProcessAsync(CancellationToken ct)
    {
        await foreach (var evt in _channel.Reader.ReadAllAsync(ct))
        {
            using var scope = _scopeFactory.CreateScope();
            var handler = scope.ServiceProvider.GetRequiredService<IOrderHandler>();
            try
            {
                await handler.HandleAsync(evt);
            }
            catch (Exception ex)
            {
                Console.WriteLine($"订单处理失败 {evt.OrderId}: {ex.Message}");
            }
        }
    }
}

// 或：Parallel.ForEachAsync 控制并发度
await Parallel.ForEachAsync(
    queue.ReadAllAsync(stoppingToken),
    new ParallelOptions { MaxDegreeOfParallelism = 8, CancellationToken = stoppingToken },
    async (evt, ct) =>
    {
        using var scope = _scopeFactory.CreateScope();
        await scope.ServiceProvider.GetRequiredService<IOrderHandler>()
            .HandleAsync(evt);
    });

// 生产者：发布订单事件
var factory = new ConnectionFactory { HostName = "rabbitmq" };
await using var conn = await factory.CreateConnectionAsync();
await using var channel = await conn.CreateChannelAsync();

await channel.ExchangeDeclareAsync("shop.events",
    type: ExchangeType.Topic, durable: true);
await channel.QueueDeclareAsync("orders.created",
    durable: true, exclusive: false, autoDelete: false);
await channel.QueueBindAsync("orders.created", "shop.events", "order.created");

var body = JsonSerializer.SerializeToUtf8Bytes(new OrderCreatedEvent(orderId));
await channel.BasicPublishAsync(
    exchange: "shop.events",
    routingKey: "order.created",
    body: body,
    mandatory: false);

// 消费者：BackgroundService 内确认
public class OrderCreatedConsumer(IConnectionFactory factory) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await using var conn = await factory.CreateConnectionAsync();
        await using var channel = await conn.CreateChannelAsync();
        await channel.BasicQosAsync(prefetchCount: 10);   // 预取上限

        var consumer = new AsyncEventingBasicConsumer(channel);
        consumer.ReceivedAsync += async (_, ea) =>
        {
            try
            {
                var evt = JsonSerializer
                    .Deserialize<OrderCreatedEvent>(ea.Body.Span)!;
                await ProcessOrderAsync(evt);
                await channel.BasicAckAsync(ea.DeliveryTag);   // 确认成功
            }
            catch
            {
                await channel.BasicNackAsync(ea.DeliveryTag,
                    requeue: false);   // 进死信，而非无限重投
            }
        };

        await channel.BasicConsumeAsync("orders.created",
            autoAck: false, consumer: consumer);
        await Task.Delay(Timeout.Infinite, stoppingToken);
    }
}

// 生产者：Session 保证同订单消息有序
var client = new ServiceBusClient(connectionString);
var sender = client.CreateSender("orders.created");

await sender.SendMessageAsync(new ServiceBusMessage(
    JsonSerializer.Serialize(new OrderCreatedEvent(orderId)))
{
    ContentType = "application/json",
    SessionId = orderId.ToString()
});

// 消费者：BackgroundService
public class OrderProcessor(ServiceBusClient client) : BackgroundService
{
    private ServiceBusProcessor? _processor;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _processor = client.CreateProcessor("orders.created",
            new ServiceBusProcessorOptions
            {
                MaxConcurrentCalls = 8,
                AutoCompleteMessages = false,   // 手动完成
                MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5)
            });

        _processor.ProcessMessageAsync += async args =>
        {
            try
            {
                var evt = args.Message.Body.ToObjectFromJson<OrderCreatedEvent>();
                await HandleOrderAsync(evt);
                await args.CompleteMessageAsync(args.Message);   // 完成
            }
            catch
            {
                await args.DeadLetterMessageAsync(args.Message,
                    "processing-failed", "处理异常");
            }
        };

        _processor.ProcessErrorAsync += args =>
        {
            Console.WriteLine($"处理器错误: {args.Exception.Message}");
            return Task.CompletedTask;
        };

        await _processor.StartProcessingAsync(stoppingToken);
        await Task.Delay(Timeout.Infinite, stoppingToken);
    }

    public override async Task StopAsync(CancellationToken ct)
    {
        if (_processor is not null)
            await _processor.StopProcessingAsync();
        await base.StopAsync(ct);
    }
}

public class RetryPolicy
{
    private const int MaxAttempts = 5;

    public async Task<bool> TryExecuteAsync(
        Func<Task> action, ILogger log, CancellationToken ct)
    {
        for (int attempt = 1; attempt <= MaxAttempts; attempt++)
        {
            try
            {
                await action();
                return true;
            }
            catch (TransientException ex) when (attempt < MaxAttempts)
            {
                // 指数退避 + 随机抖动
                var delay = TimeSpan.FromMilliseconds(
                    100 * Math.Pow(2, attempt) + Random.Shared.Next(0, 200));
                log.LogWarning("第 {Attempt} 次重试：{Message}", attempt, ex.Message);
                await Task.Delay(delay, ct);
            }
        }
        return false;   // 重试耗尽，交给上层转死信
    }
}

public class OrderHandler(OrderDbContext db, IDistributedCache cache)
{
    public async Task HandleAsync(OrderCreatedEvent evt)
    {
        var dedupKey = $"dedup:order:{evt.OrderId}";

        // 分布式锁 + 已处理标记，双重去重
        await using var lockHandle = await _lock.AcquireAsync(dedupKey, ...);
        if (lockHandle is null) return;               // 他人正在处理
        if (await cache.GetAsync(dedupKey) is not null) return;   // 已处理过
        if (await db.Orders.FindAsync(evt.OrderId) is not null) return;   // 已就位

        db.Orders.Add(new Order { Id = evt.OrderId, ... });
        await db.SaveChangesAsync();

        await cache.SetAsync(dedupKey, [1],
            new DistributedCacheEntryOptions
            { AbsoluteExpirationRelativeToNow = TimeSpan.FromDays(1) });
    }
}

// 监控：积压与处理耗时指标
public class QueueMonitor(MessageMetrics metrics) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(10));
        while (await timer.WaitForNextTickAsync(stoppingToken))
        {
            var depth = await QueryQueueDepthAsync();   // RabbitMQ 管理 API
            metrics.ObserveQueueDepth("orders.created", depth);
        }
    }
}