ABP vNext + Redis Streams：构建实时事件驱动架构

🎉 ABP vNext + Redis Streams：构建实时事件驱动架构 🚀

利用 Redis Streams 作为轻量级、高性能的消息队列，结合 ABP 的领域事件和分布式事件总线，实现微服务间的实时异步通信。 🎯

---

---

背景与目标 🎯

在微服务架构中，服务间的异步通信需满足高吞吐、低延迟，并支持横向扩展与容错。传统消息系统（如 Kafka）运维成本较高。

Redis Streams 作为 Redis 内置日志数据结构，具备轻量、持久、高性能的消息队列能力。本文演示如何在 ABP vNext 中：

• 封装 IStreamProvider 连接 Redis Streams 🔗；
• 以 BackgroundService + Consumer Group 实现弹性横向扩展 ↔️；
• 使用 Polly + DLQ 保障高可用 🔐；
• 抽取配置并接入 health checks 与监控 📊。

---

环境与依赖 🛠️

# docker-compose.yml 示例
version: '3.8'
services:redis:image: redis:7ports: ["6379:6379"]volumes:- redis-data:/datacommand: ["redis-server","--save","","--appendonly","yes"]volumes:redis-data:

• .NET 6+
• ABP vNext 6.x
• Redis 6.0+（Streams）
• Docker & Docker Compose

---

Redis Streams 概览 📑

XADD           => 追加消息
XREADGROUP     => 读取 Consumer Group 消息
XACK           => 确认消费
XDEL           => 删除已确认消息
XINFO GROUPS   => 查询消费组
XCLAIM         => 转移待处理消息

Redis Streams 支持持久化、Consumer Group、消息按 ID 追踪。

---

🏗️ 系统架构流程图

---

在 ABP 中封装 IStreamProvider 📦

RedisStreamProvider 实现

// 类定义与依赖注入
public class RedisStreamProvider : IStreamProvider, ITransientDependency
{private readonly IConnectionMultiplexer _redis;private readonly JsonSerializerOptions _serializerOptions;private readonly ILogger<RedisStreamProvider> _logger;private readonly IServiceProvider _services;public RedisStreamProvider(IConnectionMultiplexer redis,JsonSerializerOptions serializerOptions,ILogger<RedisStreamProvider> logger,IServiceProvider services){_redis = redis;_serializerOptions = serializerOptions;_logger = logger;_services = services;}// PublishAsync 实现public async Task PublishAsync<T>(string streamName, T message){var db = _redis.GetDatabase();var json = JsonSerializer.Serialize(message, _serializerOptions);await db.StreamAddAsync(streamName, new NameValueEntry[] { new NameValueEntry("data", json) });}// Subscribe 由 BackgroundService 调用，不建议在此处使用 Task.Run
}

---

注册与配置 🔧

// Startup / Program.cs
builder.Services.AddStackExchangeRedisConnection(o => o.Configuration = "localhost:6379");
builder.Services.AddSingleton<IConnectionMultiplexer>(sp =>ConnectionMultiplexer.Connect(sp.GetRequiredService<RedisConfiguration>().ConnectionString));
builder.Services.AddSingleton(new JsonSerializerOptions
{PropertyNamingPolicy = JsonNamingPolicy.CamelCase,DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull
});
builder.Services.AddTransient<IStreamProvider, RedisStreamProvider>();
// 注入配置
builder.Services.Configure<RedisStreamOptions>(configuration.GetSection("Redis:StreamSettings"));
// 注册 BackgroundService
builder.Services.AddHostedService<RedisStreamBackgroundService>();

---

使用 Consumer Group 实现横向 扩展 ↔️

# 幂等创建消费组
redis-cli --raw XINFO GROUPS mystream | grep -q 'mygroup' || redis-cli XGROUP CREATE mystream mygroup $ MKSTREAM

• 多实例启动时，注入不同 consumerName，自动均摊消息；
• 闲置消息可通过 XCLAIM 转移。

---

领域事件发布订阅最佳实践 📜

public class OrderCreatedEvent : EventData
{ public Guid OrderId { get; set; } }public class OrderDomainService : DomainService
{private readonly IStreamProvider _stream;public OrderDomainService(IStreamProvider stream) => _stream = stream;public async Task CreateOrderAsync(OrderDto dto){// 业务逻辑await _stream.PublishAsync("order-stream", new OrderCreatedEvent { OrderId = dto.Id });}
}

---

异常重试与死信队列策略 🔄

// 在 BackgroundService 中
var options = // IOptions<RedisStreamOptions>
var policy = Policy.Handle<Exception>().WaitAndRetryAsync(retryCount: options.Value.RetryCount,sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),onRetry: (ex, time) => _logger.LogWarning(ex, "Retry after {Delay}", time));await policy.ExecuteAsync(async () =>
{// 处理消息
});// DLQ 处理
await db.StreamAddAsync(options.Value.DlqStream,new NameValueEntry[] { /* data, metadata ... */ });

---

配置示例 🛠️

// appsettings.json
{"Redis": {"ConnectionString": "localhost:6379","StreamSettings": {"GroupName": "mygroup","DlqStream": "dlq-stream","RetryCount": 3,"BatchSize": 10}}
}

---

可观察性 & 监控 📊

• 健康检查：实现 IHealthCheck，检查 Redis 连接和消费组状态；
• 指标打点：使用 OpenTelemetry 或 Prometheus 收集：
  • 消息处理耗时
  • 失败率
  • Pending 队列长度

---

测试策略 🧪

1. 单元测试：Mock IConnectionMultiplexer、IDatabase，验证序列化、StreamAdd、重试逻辑；
2. 集成测试：使用 Testcontainers Redis，验证端到端 Publish → Consume → Ack 流程；
3. 模拟故障：断开 Redis 或注入异常，确保 DLQ 与重试策略生效。

---