一、Redis Stream存储对象：消息流中的结构化数据

Redis Stream作为5.0版本引入的日志型数据结构，天然适合存储序列化后的对象数据。其核心优势在于时间有序性与消费者组模式，为对象存储提供了消息队列与持久化结合的能力。

1.1 对象序列化方案

存储对象时，需将对象转换为Stream可识别的格式。常见方案包括：

• JSON序列化：通用性强，但解析开销较高。例如存储用户行为日志：
  ```python
  import json
  import redis

r = redis.Redis()
user_event = {
“user_id”: “u1001”,
“action”: “click”,
“timestamp”: 1625097600,
“metadata”: {“page”: “home”, “element”: “btn_submit”}
}
r.xadd(“user_events”, {“data”: json.dumps(user_event)})

- **MessagePack/Protocol Buffers**：二进制格式，空间效率提升30%-50%。适用于高频写入场景。
- **自定义序列化**：针对特定对象结构优化，如游戏服务器存储玩家状态：
```python
def serialize_player(player):
    return f"{player.id}:{player.level}:{player.exp}"
player_data = serialize_player(Player(id="p001", level=10, exp=1500))
r.xadd("player_states", {"player": player_data})

1.2 性能优化策略

• 批量写入：使用XADD管道模式，减少网络往返：

  pipe = r.pipeline()
  for i in range(100):
    pipe.xadd("sensor_data", {"temp": 25+i, "humidity": 60-i})
  pipe.execute()

• 合理设置Stream长度：通过XTRIM控制内存占用：

  r.xtrim("user_events", maxlen=10000, approx=True)  # 近似保留1万条

• 消费者组负载均衡：多消费者并行处理，避免单点瓶颈：

  # 创建消费者组
  XGROUP CREATE user_events mygroup $ MKSTREAM
  # 消费者1读取
  XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS user_events >

1.3 典型应用场景

• 实时日志分析：存储微服务调用链对象，支持后续聚合查询。
• 物联网设备上报：存储传感器对象，结合消费者组实现异常检测。
• 订单状态跟踪：记录订单对象变更历史，支持审计回溯。

二、Redis集合存储对象集合：高效去重与快速查询

Redis集合（Set）与有序集合（Sorted Set）为对象集合的存储提供了去重与范围查询能力，适用于需要快速判断成员存在性的场景。

2.1 对象集合存储方案

• 哈希字段作为集合元素：将对象ID或唯一标识存入集合：

  # 存储用户关注列表
  user_id = "u1001"
  followed_users = ["u2001", "u2002", "u2003"]
  r.sadd(f"user:{user_id}:follows", *followed_users)

• 有序集合存储带权重的对象：如排行榜场景：

  # 存储游戏玩家分数
  r.zadd("game_leaderboard", {"p001": 1500, "p002": 1200, "p003": 1800})

2.2 集合操作优化

• 批量操作：使用SADD/SREM的多参数版本减少命令次数：

  new_follows = ["u2004", "u2005"]
  r.sadd(f"user:{user_id}:follows", *new_follows)

• 交并差运算：高效实现复杂查询：

  # 找出同时关注A和B的用户
  common_follows = r.sinter("userfollows", "userfollows")

• 扫描大集合：使用SSCAN避免阻塞：

  cursor = 0
  while True:
    cursor, items = r.sscan("large_set", cursor=cursor, count=100)
    if not items:
        break
    # 处理items

2.3 实际应用案例

• 社交关系链：存储用户好友、关注、黑名单等集合，支持快速关系查询。
• 标签系统：为文章、商品打标签，通过集合运算实现推荐：

  # 找出同时带有"科技"和"AI"标签的文章
  article_ids = r.sinter("tag:tech", "tag:ai")

• 分布式锁：利用集合的唯一性实现简单锁机制（需配合其他结构确保安全性）。

三、Stream与集合的协同使用

实际场景中，Stream与集合常结合使用：

• Stream存储对象变更历史，集合存储当前状态。例如电商订单：
  • Stream：orders:stream 存储订单创建、支付、发货等事件。
  • 集合：orders:active 存储未完成订单ID，orders:completed 存储已完成订单ID。
• 数据同步：通过Stream消费更新集合。例如缓存预热：

  def process_stream_event(event):
    data = json.loads(event["data"])
    if data["type"] == "product_update":
        # 更新集合中的产品ID
        r.sadd("products:to_cache", data["product_id"])
        # 同时将完整对象存入Hash
        r.hset(f"product:{data['product_id']}", mapping=data)

四、最佳实践建议

1. 序列化选择：根据读写频率选择格式。高频写入场景优先MessagePack，需要人类可读时选JSON。
2. 内存控制：为Stream设置合理的MAXLEN，定期清理过期集合。
3. 消费者组设计：根据业务并发量划分消费者组，避免消息堆积。
4. 监控告警：监控Stream长度、集合元素数，设置阈值告警。
5. 备份策略：对关键Stream启用AOF持久化，定期RDB备份集合数据。

五、总结

Redis Stream与集合为对象存储提供了互补的解决方案：Stream擅长处理有序的消息流和历史追溯，集合则专注于当前状态的快速查询与去重。通过合理组合两者，开发者可以构建出高性能、低延迟的实时数据处理系统。在实际应用中，需根据业务特点选择序列化方式、优化操作命令，并建立完善的监控机制，以确保系统的稳定性和可扩展性。