Redis

Redis 扩展提供了对 Redis 的支持，适用于需要高性能和低延迟的场景。它实现了以下接口：

CommandBus - 命令总线
DomainEventBus - 领域事件总线
StateEventBus - 状态事件总线
EventStore - 事件存储
SnapshotStore - 快照存储
PrepareKey - 预分配 Key

架构概述

安装

Gradle(Kotlin)Gradle(Groovy)Maven

kotlin

implementation("me.ahoo.wow:wow-redis")
implementation("org.springframework.boot:spring-boot-starter-data-redis-reactive")

groovy

implementation 'me.ahoo.wow:wow-redis'
implementation 'org.springframework.boot:spring-boot-starter-data-redis-reactive'

xml

<dependency>
    <groupId>me.ahoo.wow</groupId>
    <artifactId>wow-redis</artifactId>
    <version>${wow.version}</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>

配置

配置类：RedisProperties
前缀：wow.redis.

名称	数据类型	说明	默认值
`enabled`	`Boolean`	是否启用	`true`
`message-bus.recovery.enabled`	`Boolean`	是否恢复被遗留的 pending 消息	`true`
`message-bus.recovery.min-idle-time`	`Duration`	触发恢复前的最小空闲时间	`5m`
`message-bus.recovery.interval`	`Duration`	pending 消息扫描间隔	`30s`
`message-bus.recovery.batch-size`	`Long`	每页 `XPENDING` 最大记录数	`100`

YAML 配置样例

yaml

spring:
  data:
    redis:
      host: localhost
      port: 6379
      password: your-password

wow:
  command:
    bus:
      type: redis
  event:
    bus:
      type: redis
  eventsourcing:
    store:
      storage: redis
    snapshot:
      storage: redis
    state:
      bus:
        type: redis
  redis:
    enabled: true
    message-bus:
      recovery:
        enabled: true
        min-idle-time: 5m
        interval: 30s
        batch-size: 100

命令总线

Redis 命令总线使用 Redis Streams 实现消息传递：

Stream 命名规则

{prefix}{contextName}.{aggregateName}.command

示例：wow.order-service.order.command

消费者组

每个处理器对应一个消费者组：

{contextName}.{processorName}

Pending 消息恢复

Redis 消息总线会在 min-idle-time 后恢复旧 consumer 遗留在 Pending Entries List（PEL）中的记录。恢复流程使用 consumer lease、有界 XPENDING 扫描和原子 XCLAIM。恢复链路失败不会终止实时消费，而会在下一轮扫描时重试。

消息语义仍为 at-least-once，因此 handler 必须幂等；min-idle-time 必须覆盖最长 handler 执行、重试和优雅停机时间。恢复默认开启。仅应在临时回滚时设置 wow.redis.message-bus.recovery.enabled=false；关闭后会恢复旧行为，使被遗留的 PEL 记录可能永久滞留。

缺少 msg 字段或 JSON 非法的记录不会再终止 consumer。此类记录保留在 PEL 中供人工诊断，同时输出不含 payload 的错误日志和 RedisMessageBusObservation.RecordDecodeFailed 观测事件。应用可注册非阻塞的 RedisMessageBusObserver Bean 接入指标或告警。

事件总线

领域事件 Stream

{prefix}{contextName}.{aggregateName}.event

状态事件 Stream

{prefix}{contextName}.{aggregateName}.state

事件存储

Redis 事件存储使用分桶的 Redis Cluster hash tag，使事件流追加、请求幂等性和聚合 ID 扫描索引可以在同一个 Lua 脚本中原子更新。

数据结构

事件流 ZSET Key: {{contextAlias}.{aggregateName}:es:{bucket}}:{aggregateId}@{tenantId}
Score: {version}
Member: {eventStreamJson}

请求 ID SET Key: {{contextAlias}.{aggregateName}:es:{bucket}}:{aggregateId}@{tenantId}:req_idx
Member: {requestId}

聚合 ID ZSET Key: {{contextAlias}.{aggregateName}:es:{bucket}}:ids
Score: 0
Member: {aggregateId}\u0000{tenantId}

请求幂等性

请求 ID 存储在上面的同分桶 SET Key 中。

聚合 ID 扫描

EventStore.scanAggregateId 会扫描分桶的聚合 ID 索引，并按字典序合并结果。由于 aggregateId 全局唯一，scanner 为每个聚合存储一个 member，并直接从 ZSET member 中解码 tenantId。

快照存储

快照使用 Hash 结构存储：

Key: {prefix}{contextName}.{aggregateName}:{aggregateId}:snapshot
Value: {snapshotJson}

预分配 Key

PrepareKey 使用 String 结构：

Key: {prefix}prepare:{keyName}:{key}
Value: {preparedValue}
TTL: 根据配置或永久

连接池配置

yaml

spring:
  data:
    redis:
      lettuce:
        pool:
          min-idle: 8
          max-idle: 16
          max-active: 32
          max-wait: 1000ms

参数	说明	推荐值
`min-idle`	最小空闲连接	8
`max-idle`	最大空闲连接	16
`max-active`	最大活动连接	32
`max-wait`	最大等待时间	1000ms

集群配置

yaml

spring:
  data:
    redis:
      cluster:
        nodes:
          - redis-node-1:6379
          - redis-node-2:6379
          - redis-node-3:6379
        max-redirects: 3
      lettuce:
        cluster:
          refresh:
            adaptive: true
            period: 30s

哨兵配置

yaml

spring:
  data:
    redis:
      sentinel:
        master: mymaster
        nodes:
          - sentinel-1:26379
          - sentinel-2:26379
          - sentinel-3:26379

性能优化

Pipeline 批量操作

Redis 扩展自动使用 Pipeline 优化批量操作，减少网络往返次数。

内存优化

合理设置 TTL：为临时数据设置过期时间
压缩存储：启用数据压缩减少内存占用
监控内存：定期检查内存使用情况

建议配置

yaml

# Redis 服务端配置建议
maxmemory 4gb
maxmemory-policy allkeys-lru
tcp-keepalive 300
timeout 0

故障排查

常见问题

1. 连接超时

org.springframework.data.redis.RedisConnectionFailureException

解决方案：

检查 Redis 服务状态
验证网络连通性
调整连接超时配置

2. 内存不足

OOM command not allowed when used memory > 'maxmemory'

解决方案：

增加 Redis 内存限制
配置合理的过期策略
清理不必要的数据

3. Stream 消费延迟

解决方案：

增加消费者数量
优化消息处理逻辑
检查 Redis 服务性能

监控指标

建议监控以下 Redis 指标：

指标	说明	告警阈值
`used_memory`	内存使用	> 80% maxmemory
`connected_clients`	连接数	> 1000
`blocked_clients`	阻塞客户端数	> 10
`keyspace_hits/misses`	缓存命中率	< 90%

完整配置示例

yaml

spring:
  data:
    redis:
      host: localhost
      port: 6379
      password: your-password
      database: 0
      lettuce:
        pool:
          min-idle: 8
          max-idle: 16
          max-active: 32
          max-wait: 1000ms
        shutdown-timeout: 100ms

wow:
  command:
    bus:
      type: redis
      local-first:
        enabled: true
  event:
    bus:
      type: redis
      local-first:
        enabled: true
  eventsourcing:
    store:
      storage: redis
    snapshot:
      enabled: true
      strategy: all
      storage: redis
    state:
      bus:
        type: redis
        local-first:
          enabled: true
  redis:
    enabled: true

最佳实践

启用 LocalFirst 模式：本地消息优先处理，减少网络延迟
使用集群模式：生产环境使用 Redis 集群保证高可用和扩展性
合理配置连接池：根据并发量配置适当的连接池大小
监控内存使用：定期监控 Redis 内存使用，避免 OOM
启用持久化：配置 RDB 或 AOF 持久化防止数据丢失

Redis ​

架构概述 ​

安装 ​

配置 ​

命令总线 ​

Stream 命名规则 ​

消费者组 ​

Pending 消息恢复 ​

事件总线 ​

领域事件 Stream ​

状态事件 Stream ​

事件存储 ​

数据结构 ​

请求幂等性 ​

聚合 ID 扫描 ​

快照存储 ​

预分配 Key ​

连接池配置 ​

集群配置 ​

哨兵配置 ​

性能优化 ​

Pipeline 批量操作 ​

内存优化 ​

建议配置 ​

故障排查 ​

常见问题 ​

1. 连接超时 ​

2. 内存不足 ​

3. Stream 消费延迟 ​

监控指标 ​

完整配置示例 ​

最佳实践 ​

Redis

架构概述

安装

配置

命令总线

Stream 命名规则

消费者组

Pending 消息恢复

事件总线

领域事件 Stream

状态事件 Stream

事件存储

数据结构

请求幂等性

聚合 ID 扫描

快照存储

预分配 Key

连接池配置

集群配置

哨兵配置

性能优化

Pipeline 批量操作

内存优化

建议配置

故障排查

常见问题

1. 连接超时

2. 内存不足

3. Stream 消费延迟

监控指标

完整配置示例

最佳实践