1
public async Task<DeviceConfigDto?> GetConfigWithMutexAsync(string deviceId)
2
public async Task<ProductDetailDto?> GetHotProductWithMutexAsync(string productId)
3
{
4
    string cacheKey = $"cache:product:detail:{productId}";
5
    var cached = await _redis.StringGetAsync(cacheKey);
6
    if (cached.HasValue) return JsonSerializer.Deserialize<ProductDetailDto>(cached!);
7

8
    // 未命中，尝试获取互斥锁 (防止瞬时万级并发打穿数据库)
9
    string lockKey = $"lock:product:detail:{productId}";
10
    string lockToken = Guid.NewGuid().ToString("N");
11

12
    // 仅持锁 5 秒
13
    if (await _redis.StringSetAsync(lockKey, lockToken, TimeSpan.FromSeconds(5), When.NotExists))
14
    {
15
        try
16
        {
17
            // 双重校验 (Double-Check)，防止排队线程重复查 DB
18
            cached = await _redis.StringGetAsync(cacheKey);
19
            if (cached.HasValue) return JsonSerializer.Deserialize<ProductDetailDto>(cached!);
20

21
            var dbData = await _db.Products.FindAsync(productId);
22
            var serialized = JsonSerializer.Serialize(dbData);
23
            await _redis.StringSetAsync(cacheKey, serialized, TimeSpan.FromMinutes(30));
24
            return dbData;
25
        }
26
        finally
27
        {
28
            // Lua 原子释放锁
29
            await ReleaseLockAsync(lockKey, lockToken);
30
        }
31
    }
32
    else
33
    {
34
        // 没抢到锁，等待 50ms 重新递归读取缓存
35
        await Task.Delay(50);
36
        return await GetHotProductWithMutexAsync(productId);
37
    }
38
}

1
public async Task UpdateProductPriceAsync(string productId, decimal newPrice)
2
{
3
    string cacheKey = $"cache:product:detail:{productId}";
4

5
    // 1. 先删缓存
6
    await _redis.KeyDeleteAsync(cacheKey);
7

8
    // 2. 更新数据库
9
    await _db.UpdatePriceAsync(productId, newPrice);
10

11
    // 3. 异步延迟双删 (等待主从复制与并发读完成)
12
    _ = Task.Run(async () =>
13
    {
14
        await Task.Delay(500); // 延迟 500ms
15
        await _redis.KeyDeleteAsync(cacheKey);
16
    });
17
}

1
sequenceDiagram
2
    autonumber
3
    participant App as 订单服务
4
    participant DelayQueue as 延迟等待队列 (设置 TTL = 30min, 无消费者)
5
    participant DLX as 死信交换机 (DLX)
6
    participant TargetQueue as 真正消费队列 (订单关闭队列)
7
    participant Consumer as 取消订单消费者
8

9
    App->>DelayQueue: 发送订单消息 (带有 30 分钟过期时间)
10
    Note over DelayQueue: 消息在队列中等待 30 分钟... 自然超时变为死信！
11
    DelayQueue->>DLX: 消息自动路由至 DLX
12
    DLX->>TargetQueue: 转发至真正处理队列
13
    TargetQueue->>Consumer: 消费者触发：核对状态并释放库存

1
public class OrderCreatedConsumer : AsyncDefaultBasicConsumer
2
{
3
    private readonly IServiceProvider _serviceProvider;
4
    private readonly IDatabase _redis;
5

6
    public OrderCreatedConsumer(IModel channel, IServiceProvider serviceProvider, IConnectionMultiplexer redis)
7
        : base(channel)
8
    {
9
        _serviceProvider = serviceProvider;
10
        _redis = redis.GetDatabase();
11
    }
12

13
    public override async Task HandleBasicDeliver(
14
        string consumerTag,
15
        ulong deliveryTag,
16
        bool redelivered,
17
        string exchange,
18
        string routingKey,
19
        IBasicProperties properties,
20
        ReadOnlyMemory<byte> body)
21
    {
22
        // 1. 提取消息全局唯一标识 (MessageId)
23
        string messageId = properties.MessageId;
24
        if (string.IsNullOrWhiteSpace(messageId))
25
        {
26
            Model.BasicReject(deliveryTag, requeue: false); // 无标识垃圾报文丢弃
27
            return;
28
        }
29

30
        // 2. 利用 Redis SETNX 进行原子去重校验 (防重有效期 24 小时)
31
        string deduplicationKey = $"mq:consumed:{messageId}";
32
        bool isFirstTime = await _redis.StringSetAsync(
33
            deduplicationKey,
34
            "1",
35
            TimeSpan.FromHours(24),
36
            When.NotExists);
37

38
        if (!isFirstTime)
39
        {
40
            // 说明此消息之前已经成功消费过，直接 Ack 短路，避免重复执行业务
41
            Model.BasicAck(deliveryTag, multiple: false);
42
            return;
43
        }
44

45
        try
46
        {
47
            // 3. 执行真正的业务操作
48
            using var scope = _serviceProvider.CreateScope();
49
            var orderService = scope.ServiceProvider.GetRequiredService<IOrderService>();
50

51
            var payload = JsonSerializer.Deserialize<OrderCreatedEvent>(body.Span);
52
            await orderService.ProcessOrderCreatedAsync(payload!);
53

54
            // 4. 业务成功，手动 Ack
55
            Model.BasicAck(deliveryTag, multiple: false);
56
        }
57
        catch (Exception ex)
58
        {
59
            // 业务异常：删除 Redis 去重标记，并拒收消息或进入重试死信队列
60
            await _redis.KeyDeleteAsync(deduplicationKey);
61
            Model.BasicNack(deliveryTag, multiple: false, requeue: redelivered == false);
62
        }
63
    }
64
}

1
// ❌ 严重隐患代码：
2
await _db.SaveChangesAsync();
3
await _mqPublisher.PublishAsync(eventMsg); // 若此时网络闪断、RabbitMQ 宕机或进程断电，消息永远丢失，下游永远无法收到通知！

1
flowchart TD
2
    subgraph 业务服务本地事务 (ACID 强保证)
3
        S1[写入业务数据: Orders] --> Tx[同一个数据库本地事务]
4
        S2["写入待发事件: OutboxMessages\n(Status: Pending)"] --> Tx
5
    end
6
    Tx -->|提交成功| DB[(应用数据库)]
7

8
    subgraph 后台异步轮询发布器 (Worker)
9
        Worker[BackgroundService / Debezium] -->|扫描 Pending 消息| DB
10
        Worker -->|成功投递| RMQ[RabbitMQ Broker]
11
        Worker -->|更新状态为 Published| DB
12
    end