Redis 高级数据结构实战：从 Bitmap 到 Stream 的落地指南

为什么需要高级数据结构？

很多开发者对 Redis 的认知还停留在 String、Hash、List、Set、ZSet 这五种基础类型上。但在实际生产环境中，我们常遇到一些特殊需求：

• 统计每天上亿用户的活跃情况，但内存有限
• 计算附近的人，且要求毫秒级响应
• 实现一个轻量级消息队列，支持多消费者组
• 批量执行命令，减少网络往返

这些场景如果只用基础类型，要么实现复杂，要么性能堪忧。Redis 提供了一系列高级数据结构，专门解决这类问题。本文将通过五个实战案例，带你彻底掌握它们。

> 本文代码基于 Python 3.8+ 和 redis-py 4.x，所有示例均可在本地 Redis 中直接运行。

环境准备

确保已安装 Redis 和 Python 依赖：

# 安装 Redis（略，请参考官方文档）
# 安装 Python 客户端
pip install redis

启动 Redis 服务后，用以下代码测试连接：

import redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
print(r.ping())  # 输出 True 表示连接成功

1. Bitmap：海量布尔统计的利器

场景

假设你运营一个日活千万的 App，需要统计每个用户的每日签到情况，并支持快速查询某用户某天是否签到、某天总签到人数、连续签到天数等。用 Set 存储，每个用户 ID 占 8 字节，一千万用户就是 80MB，一个月就是 2.4GB。而 Bitmap 每个用户只占 1 bit，内存节省 64 倍。

核心命令

• SETBIT key offset value：设置某一位为 0 或 1
• GETBIT key offset：获取某一位的值
• BITCOUNT key [start end]：统计为 1 的位数
• BITOP operation destkey key [key ...]：位运算（AND/OR/XOR/NOT）

实战：用户签到系统

import redis
from datetime import date

r = redis.Redis(decode_responses=True)

def sign_in(user_id, day=None):
    """用户签到，day 格式 YYYYMMDD，默认为今天"""
    if day is None:
        day = date.today().strftime('%Y%m%d')
    key = f'sign:{day}'
    # 用户 ID 作为 offset，假设用户 ID 从 1 开始
    r.setbit(key, user_id, 1)

def is_signed(user_id, day):
    """检查某天是否签到"""
    key = f'sign:{day}'
    return r.getbit(key, user_id) == 1

def count_signs(day):
    """统计某天签到总人数"""
    key = f'sign:{day}'
    return r.bitcount(key)

def continuous_days(user_id, end_day):
    """计算截至 end_day 的连续签到天数（需从当天向前检查）"""
    from datetime import datetime, timedelta
    end = datetime.strptime(end_day, '%Y%m%d')
    days = 0
    while True:
        day_str = end.strftime('%Y%m%d')
        if is_signed(user_id, day_str):
            days += 1
            end -= timedelta(days=1)
        else:
            break
    return days

# 测试
sign_in(1001, '20250301')
sign_in(1001, '20250302')
sign_in(1001, '20250303')
print(is_signed(1001, '20250302'))  # True
print(count_signs('20250303'))      # 1
print(continuous_days(1001, '20250303'))  # 3

注意：Bitmap 的 offset 最大为 2^32-1（512MB），如果用户 ID 稀疏且范围极大，会造成大量内存浪费。此时建议用 Hash 分片或改用其他方案。

💡 最佳实践：对于按月统计，可以用 BITOP AND 计算连续签到用户。例如 BITOP AND dest sign:20250301 sign:20250302 得到两天都签到的用户位图。

2. HyperLogLog：不精确但极省内存的基数统计

场景

统计网站每日 UV（独立访客），如果使用 Set 存储用户 ID，一亿用户需要约 800MB。而 HyperLogLog 只需要 12KB，误差率仅 0.81%。对于不需要精确值的统计场景，这是绝佳选择。

核心命令

• PFADD key element [element ...]：添加元素
• PFCOUNT key [key ...]：返回基数估计值
• PFMERGE destkey sourcekey [sourcekey ...]：合并多个 HyperLogLog

实战：UV 统计

def add_uv(day, user_id):
    key = f'uv:{day}'
    r.pfadd(key, user_id)

def get_uv(day):
    return r.pfcount(f'uv:{day}')

def get_uv_range(days):
    """统计多天总 UV（去重）"""
    keys = [f'uv:{day}' for day in days]
    return r.pfcount(*keys)

# 模拟添加 10000 个用户
for i in range(10000):
    add_uv('20250303', f'user_{i}')
print(get_uv('20250303'))  # 约 10000，可能有微小误差

注意：HyperLogLog 不存储元素本身，只用于计数，且无法判断某个元素是否存在。如果需要同时判断存在性，应结合 Bloom Filter（Redis 模块）或 Set。

💡 最佳实践：HyperLogLog 适合统计大规模去重计数，如 UV、搜索关键词去重数。但要注意它不保证精确，如果业务要求 100% 准确，请使用 Set 或 Sorted Set。

3. GEO：地理位置计算

场景

“附近的人”、“附近的餐厅”是常见需求。Redis 的 GEO 基于 Sorted Set 实现，支持存储经纬度、计算距离、按半径查询。

核心命令

• GEOADD key longitude latitude member：添加位置
• GEODIST key member1 member2 [unit]：计算距离
• GEOSEARCH key FROMMEMBER member BYRADIUS radius unit：半径搜索
• GEOPOS key member：获取坐标

实战：附近商家查询

def add_shop(shop_id, lng, lat):
    r.geoadd('shops', (lng, lat, shop_id))

def nearby_shops(lng, lat, radius_km, count=10):
    """查询附近商家，按距离排序"""
    results = r.geosearch(
        'shops',
        longitude=lng,
        latitude=lat,
        radius=radius_km,
        unit='km',
        sort='ASC',
        count=count,
        withdist=True,
        withcoord=True
    )
    return results

# 添加几个北京地标
add_shop('tiananmen', 116.397, 39.909)
add_shop('wangfujing', 116.410, 39.915)
add_shop('sanlitun', 116.455, 39.933)

# 查询天安门附近 5km 的商家
print(nearby_shops(116.397, 39.909, 5))
# 输出类似：[['tiananmen', 0.0, [116.397, 39.909]], ['wangfujing', 1.2, [116.410, 39.915]], ...]

注意：GEO 的底层是 ZSet，所以可以用 ZREM 删除元素。另外，GEO 的精度受限于 geohash 编码，距离计算可能有微小误差。

💡 最佳实践：对于海量位置数据，建议按城市或区域分 key，避免单个 key 过大。同时，可以结合 GEORADIUS 的 STORE 选项将结果存入新 key 供后续使用。

4. Stream：轻量级消息队列

场景

需要实现一个简单的消息队列，支持多消费者、消息持久化、ACK 机制，但又不想引入 Kafka 或 RabbitMQ。Redis 5.0 引入的 Stream 正是为此设计。

核心命令

• XADD key field value ...：添加消息， 表示自动生成 ID
• XREADGROUP GROUP group consumer [COUNT n] [BLOCK ms] STREAMS key >：消费者组读取
• XACK key group id：确认消息
• XPENDING key group：查看待处理消息
• XCLAIM key group consumer min-idle-time id：转移消息所有权

实战：订单处理队列

import time

def produce_order(order_id):
    msg_id = r.xadd('orders', {'order_id': order_id, 'status': 'pending'})
    return msg_id

def consume_orders(group, consumer, count=1, block=5000):
    """从消费者组读取消息，block 毫秒超时"""
    # 创建消费者组（如果不存在）
    try:
        r.xgroup_create('orders', group, id='0', mkstream=True)
    except redis.exceptions.ResponseError as e:
        if 'BUSYGROUP' not in str(e):
            raise
    
    messages = r.xreadgroup(group, consumer, {'orders': '>'}, count=count, block=block)
    if not messages:
        return []
    
    # 处理消息
    results = []
    for stream, msgs in messages:
        for msg_id, fields in msgs:
            # 模拟业务处理
            print(f"处理订单 {fields['order_id']}")
            results.append((msg_id, fields))
            # 处理成功后 ACK
            r.xack('orders', group, msg_id)
    return results

# 生产 5 条消息
for i in range(5):
    produce_order(f'ORDER_{i}')

# 消费者 A 消费
consume_orders('order_group', 'consumer_A', count=3)
# 消费者 B 消费剩余
consume_orders('order_group', 'consumer_B', count=3)

注意：XREADGROUP 使用 > 表示只读取新消息，已投递但未 ACK 的消息不会再次投递。如果需要重新处理失败消息，需用 XPENDING + XCLAIM 或 XAUTOCLAIM。

💡 最佳实践：

• 为消息设置最大长度：XADD orders MAXLEN ~ 10000 * ... 避免内存无限增长。
• 使用 XAUTOCLAIM 自动转移超时未 ACK 的消息，实现故障恢复。
• 消费者组名称和消费者名称应有意义，便于监控。

5. Pipeline：批量执行与性能优化

场景

当需要执行大量 Redis 命令时，每个命令一次网络往返（RTT）会成为瓶颈。Pipeline 允许一次性发送多条命令，减少网络开销。

实战：批量插入与查询

def batch_set(keys_values):
    """批量设置键值对"""
    pipe = r.pipeline(transaction=False)  # 非事务模式，性能更高
    for key, value in keys_values:
        pipe.set(key, value)
    return pipe.execute()

def batch_get(keys):
    """批量获取"""
    pipe = r.pipeline(transaction=False)
    for key in keys:
        pipe.get(key)
    return pipe.execute()

# 测试
kv = [(f'key_{i}', f'value_{i}') for i in range(1000)]
batch_set(kv)
values = batch_get([f'key_{i}' for i in range(1000)])
print(len(values))  # 1000

注意：Pipeline 不保证原子性（除非使用 transaction=True），命令按顺序执行，但中间可能插入其他客户端的命令。如果业务需要原子性，请使用 MULTI/EXEC 事务或 Lua 脚本。

💡 最佳实践：

• 控制单次 Pipeline 的命令数量（建议 500-1000 条），避免阻塞 Redis 和客户端内存溢出。
• 对于只读场景，可以结合 CLIENT REPLY OFF 进一步减少网络流量。

总结

本文介绍了 Redis 五种高级数据结构及其典型应用场景：

| 数据结构 | 适用场景 | 内存/性能特点 | |———|———|————–| | Bitmap | 布尔统计、签到 | 极省内存，支持位运算 | | HyperLogLog | 基数统计（UV） | 固定 12KB，误差 0.81% | | GEO | 地理位置 | 基于 ZSet，支持半径搜索 | | Stream | 消息队列 | 持久化、消费者组、ACK | | Pipeline | 批量操作 | 减少 RTT，提升吞吐 |

在实际项目中，往往需要组合使用。例如，用 Bitmap 记录用户签到，用 HyperLogLog 统计 UV，用 GEO 实现附近查询，用 Stream 处理异步任务，用 Pipeline 优化批量写入。

下一步学习建议：

• 深入了解 Redis 模块（如 RedisBloom、RedisJSON）
• 学习 Lua 脚本实现原子操作
• 研究 Redis 持久化（RDB/AOF）与集群方案

希望本文能帮助你在实际工作中更好地利用 Redis。如果有任何问题，欢迎留言讨论。