Crping.TaskManager.Redis 3.0.0

Crping.TaskManager.Redis

Crping.TaskManager 的 Redis Stream 扩展，提供基于 Redis Stream 的消息队列支持。

版本更新说明

3.0.0

2026年7月30日 星期三

• 从 Crping.TaskManager 拆分独立为 Crping.TaskManager.Redis
• 版本号统一更新至 3.0.0

1.0.0

2026年1月29日 星期三

• 初始版本
• 从 Crping.TaskManager 中拆分 Redis 相关功能
• 新增 IReceiverWithRedis 接口
• 新增 ReceiverWithRedis<TReceiver> 抽象基类
• 新增 RedisUtils 序列化工具类
• 修复 AcknowledgedIds 线程安全问题（从 static 改为实例字段）

功能特性

• Redis Stream 消费者组支持：基于 Redis Stream 实现可靠的消息队列消费
• 消息确认与删除：支持消息确认（ACK）和批量删除
• Pending 消息处理：自动读取未正常处理的 Pending 消息
• 对象序列化：支持对象与 Redis Stream Entry 之间的自动序列化/反序列化
• 消费者组管理：自动创建消费者组（如果不存在）

依赖

• Crping.TaskManager (>= 3.0.0)
• StackExchange.Redis (>= 3.0.17)

快速开始

1. 注册 Redis 服务

// 在 ConfigureServices 中注册 Redis 连接
services.AddSingleton<IConnectionMultiplexer>(n =>
{
    var connection = configuration.GetConnectionString("RedisCache");
    return ConnectionMultiplexer.Connect(connection!);
});

services.AddScoped(n =>
{
    var redis = n.GetRequiredService<IConnectionMultiplexer>();
    return redis.GetDatabase();
});

2. 创建 Redis 消息生产者

using Crping.TaskManager.Abstractions;
using Crping.TaskManager.Redis;

public class MyProducer : ReceiverWithRedis<MyProducer>
{
    public override Task<List<ICommand>> CreateCommands(IServiceProvider sp, ILogger log)
    {
        // 创建 10 个命令
        return CreateSimpleCommandsWithRedis<MyProducer>(10);
    }

    public override async Task Execute(IServiceProvider sp, ILogger log)
    {
        // 生产消息到 Redis Stream
        var data = new { Id = 1, Name = "Test" };
        await Cache.StreamAddAsync("my_stream", data.ToRedisNameValueEntries());
    }
}

3. 创建 Redis 消息消费者

using Crping.TaskManager.Abstractions;
using Crping.TaskManager.Redis;

public class MyConsumer : ReceiverWithRedis<MyConsumer>
{
    public override async Task<List<ICommand>> CreateCommands(IServiceProvider sp, ILogger log)
    {
        // 从 Redis Stream 读取消息
        return await GetCommandsFromRedisAsync("my_stream");
    }

    public override async Task Execute(IServiceProvider sp, ILogger log)
    {
        // 处理消息
        var model = Entry.FromRedisStreamEntry<MyModel>();
        log.Trace($"处理消息：{model.Name}");
        
        // 确认消息已处理
        await StreamAcknowledgeAsync();
    }
}

API 参考

ReceiverWithRedis<TReceiver>

抽象基类，提供 Redis Stream 消费功能。

属性

方法

RedisUtils

提供对象与 Redis Stream Entry 之间的序列化/反序列化扩展方法。

扩展方法