Qishao.StackExchangeRedis.MQ 2026.8.7

Qishao.StackExchangeRedis.MQ

Qishao.StackExchangeRedis.MQ 使用 Redis Streams 实现 Qishao.MQ.Abstractions 的发布、消费、续租和管理能力。

安装

dotnet add package Qishao.StackExchangeRedis.MQ

支持 netstandard2.1、net6.0 和 net8.0。Redis 服务端最低版本为 5.0。

注册

using Microsoft.Extensions.DependencyInjection;

services
    .AddRedisConnection(options =>
    {
        options.ConnectionString = "localhost:6379";
        options.InstanceName = "app:";
    })
    .AddMQRedisService(options =>
    {
        options.MaxDeliveryCount = 5;
        options.PublishIdempotencyWindow = TimeSpan.FromDays(1);
    });

注册后可注入：

• IMessagePublisher
• IMessageConsumer
• IMessageLeaseManager
• IMQAdministration

API 中的 Topic 是逻辑名称。Redis 物理 key 始终由连接的 InstanceName 加 Topic 组成，不接受调用方传入已拼接前缀的 key。 默认继承 Redis 连接的 InstanceName；在 AddMQRedisService 中显式配置时，以 MQ 配置值为准，显式设置为空字符串可禁用前缀。 Topic 使用 Redis Cluster hash tag 时必须采用 {tenant} 这类非空闭合格式；空、残缺或孤立的花括号会被拒绝，避免多 Key Lua 在 Cluster 中产生 CROSSSLOT。

发布

var result = await publisher.PublishAsync(
    "orders",
    new PublishMessage(
        body,
        idempotencyKey: "order-1001",
        headers: new Dictionary<string, string>
        {
            ["traceId"] = "trace-1"
        }),
    cancellationToken);

消息体写入 data 字段，Header 使用 header: 前缀。带幂等键的发布通过同槽 Lua 原子完成检查、XADD 和幂等记录写入；重复发布返回原消息 ID。

消费与续租

Redis 6.2 及以上使用 XAUTOCLAIM 接管超时 pending 消息；Redis 5.0 使用 XPENDING 和 XCLAIM。返回的 MessageEnvelope.DeliveryCount 是当前投递次数。

IMessageLeaseManager.RenewAsync 使用 Lua 原子检查 PEL 所有者。只有当前消费者仍拥有消息时才执行 XCLAIM ... JUSTID 重置空闲时间，续租不会把其他消费者的消息抢回来，也不会增加 delivery count。

Lua 通过 StackExchange.Redis 的 ScriptEvaluateAsync 执行。客户端会在首次调用时发送完整脚本，后续自动使用 EVALSHA，无需业务代码重复执行 SCRIPT LOAD。

结算

• Complete：Lua 原子校验 PEL 所有者后执行 XACK，随后尝试安全裁剪，返回 Settled 或 LeaseLost。
• Retry：Lua 原子确认消息仍属于当前消费者，不 ACK、不续租，返回 RetainedForRetry 或 LeaseLost。
• DeadLetter：先校验并续租，再幂等写入死信 Stream，保留 Body、Headers 和源消息元数据，最后按所有者执行 XACK。

超过 MaxDeliveryCount 的消息也会自动转入死信 Stream。死信 Topic 默认为原 Topic 加 :dead-letter。

配置

保留策略

主 Stream 发布时不使用 MAXLEN 或 XTRIM。只有 Handler 正常返回 Complete 且 ACK 成功后，才会检查是否超过 MaxQueueLength，每次最多扫描最旧的 100 条候选消息：

• 所有现存消费组都已投递到该消息之后；
• 该消息不在任何消费组的 PEL 中；
• 同时满足以上条件才通过 XDEL 删除精确消息 ID。

Redis 5.0 的 XINFO GROUPS 在 Lua 中按 RESP2 字段/值数组解析。字段缺失或结构异常时，本次裁剪直接停止。每个消费组的 XPENDING 最多读取“本批候选数 + 1”条，即最多 101 条；达到上限时保守地跳过本批删除。

裁剪最多处理 MaxTrimConsumerGroups 个消费组，默认 32。消费组数量超过上限时，本次直接跳过裁剪并对该 Topic 记录一次警告。Redis 5.0 的 XINFO GROUPS 不支持分页，因此获取消费组列表本身仍是 O(G)；此上限主要阻止后续每组 XPENDING 和候选消息双层检查继续放大执行时间。

Retry、异常、死信和无效消息不会触发裁剪。受 pending 或未投递消息保护时，实际 XLEN 可以暂时高于 MaxQueueLength，因此生产环境仍需监控 Stream 长度、PEL 和 Redis 内存。