Redis位图实战：用BITFIELD实现高效用户签到系统（附完整代码）

Redis位图实战：用BITFIELD实现高效用户签到系统（附完整代码）

想象一下，一个拥有百万日活的社交应用，每天需要记录海量用户的签到行为。如果为每个用户每天单独存储一条记录，不仅浪费存储空间，还会给数据库带来巨大压力。而Redis的位图（Bitmap）特性，配合BITFIELD命令，能够以极低的内存消耗实现高性能签到系统——这正是技术团队梦寐以求的解决方案。

1. 为什么位图是签到系统的理想选择

传统签到系统通常采用关系型数据库记录用户签到日期，每条记录至少包含用户ID、签到日期两个字段。假设用户ID为8字节整数，日期为3字节字符串，那么单用户单月签到数据就需11字节存储空间。对于百万级用户的应用，每月仅签到数据就需要：

1000000用户 × 30天 × 11字节 ≈ 330MB

而采用Redis位图方案后，存储空间骤降：

• 空间效率：每个用户每月签到状态仅需4字节（32位）存储
• 计算效率：位操作时间复杂度O(1)，统计计算O(N)但N极小
• 功能完备：支持连续签到判断、补签标记等复杂业务场景

下表对比两种方案的性能差异：

2. BITFIELD命令的核心优势解析

Redis的基础位图命令如SETBIT/GETBIT虽然简单易用，但在处理连续签到等场景时存在明显局限。BITFIELD命令通过三大特性彻底改变了游戏规则：

2.1 整数值存储与原子操作

BITFIELD user:1000:signed SET u1 0 1 SET u1 1 1 SET u1 2 0

这条命令原子性地完成了：

1. 第0位设置为1（第一天签到）
2. 第1位设置为1（第二天签到）
3. 第2位设置为0（第三天未签）

2.2 智能溢出控制

BITFIELD user:1000:streak OVERFLOW SAT INCRBY u8 0 1

当连续签到天数达到类型上限时，SAT策略会保持最大值不溢出，避免数值回绕导致统计错误。

2.3 多操作批处理

BITFIELD user:1000:stats GET u8 #0 # 获取当月签到天数 GET u16 #1 # 获取连续签到天数 INCRBY u8 #0 1

单次网络往返即可完成数据查询和更新，降低延迟的同时保证原子性。

3. 完整签到系统架构设计

3.1 数据模型设计

采用三层存储结构实现空间与功能的平衡：

1. 日粒度位图（核心存储）

   user:<uid>:signed:<yyyyMM> -> 31位位图

2. 连续签到计数器

   user:<uid>:streak -> 16位无符号整数

3. 月度统计缓存

   user:<uid>:stats:<yyyyMM> -> { signed_days: 8位, max_streak: 8位, reward_status: 8位 }

3.2 关键操作实现

签到API核心逻辑：

def sign_in(user_id): today = datetime.now().day - 1 # 转为0-based month_key = f"user:{user_id}:signed:{datetime.now().strftime('%Y%m')}" streak_key = f"user:{user_id}:streak" # 原子操作：检查是否已签到并更新状态 result = redis.bitfield(month_key) .get(f'u1', today) .set(f'u1', today, 1) .execute() if result[0] == 1: raise AlreadySignedError # 更新连续签到天数（带饱和溢出保护） redis.bitfield(streak_key, 'OVERFLOW', 'SAT') .incrby('u16', 0, 1) .execute() # 更新月度统计 redis.bitfield(f"user:{user_id}:stats:{datetime.now().strftime('%Y%m')}") .incrby('u8', 0, 1) # 签到天数+1 .execute()

连续签到奖励检查：

def check_streak_reward(user_id): streak = int(redis.bitfield(f"user:{user_id}:streak") .get('u16', 0) .execute()[0]) reward_status = redis.bitfield(f"user:{user_id}:stats:{datetime.now().strftime('%Y%m')}") .get('u8', 2) # 奖励状态位 .execute()[0] if streak >= 7 and not (reward_status & 0b01): grant_reward(user_id) redis.bitfield(f"user:{user_id}:stats:{datetime.now().strftime('%Y%m')}") .set('u8', 2, reward_status | 0b01) .execute()

4. 生产环境优化实践

4.1 内存压缩技巧

通过合理设计位域结构，单用户全年签到数据可压缩到仅6字节：

# 每月用4字节存储（31天+1个月份标记位） BITFIELD user:1000:2023 SET u32 0 0xFFFFFFFF # 初始化全月数据 SET u8 4 12 # 标记月份

4.2 冷热数据分离

def migrate_monthly_data(user_id): current_month = datetime.now().strftime('%Y%m') prev_month = (datetime.now() - timedelta(days=31)).strftime('%Y%m') # 迁移上月数据到RDBMS signed_data = redis.bitfield(f"user:{user_id}:signed:{prev_month}") .get('u32', 0) .execute()[0] save_to_database(user_id, prev_month, signed_data) # 清理Redis过期数据 redis.delete( f"user:{user_id}:signed:{prev_month}", f"user:{user_id}:stats:{prev_month}" )

4.3 异常处理策略

def handle_bitfield_overflow(): try: redis.bitfield('counter', 'OVERFLOW', 'FAIL') .incrby('u8', 0, 1) .execute() except ResponseError as e: if 'overflow' in str(e): send_alert('Counter overflow detected') redis.bitfield('counter', 'OVERFLOW', 'SAT') .set('u8', 0, 255) .execute()

5. 性能压测对比

在4核8G的Redis实例上，使用redis-benchmark工具测试：

内存占用对比测试结果：

6. 完整实现代码示例

import redis from datetime import datetime class BitmapSignSystem: def __init__(self, host='localhost', port=6379): self.redis = redis.StrictRedis(host=host, port=port, decode_responses=False) def sign_in(self, user_id: int) -> bool: """用户签到返回是否首次签到""" today = datetime.now().day - 1 month_key = f"user:{user_id}:signed:{datetime.now().strftime('%Y%m')}" streak_key = f"user:{user_id}:streak" # 原子操作：检查并设置签到状态 try: result = self.redis.bitfield(month_key) .get(f'u1', today) .set(f'u1', today, 1) .execute() if result[0] == 1: return False # 更新连续签到天数 self.redis.bitfield(streak_key, 'OVERFLOW', 'SAT') .incrby('u16', 0, 1) .execute() # 更新月度统计 self.redis.bitfield(f"user:{user_id}:stats:{datetime.now().strftime('%Y%m')}") .incrby('u8', 0, 1) # 签到总天数 .execute() return True except redis.ResponseError as e: raise SystemError(f"Redis operation failed: {str(e)}") def get_sign_status(self, user_id: int, year: int, month: int) -> dict: """获取指定月份签到状态""" month_key = f"user:{user_id}:signed:{year}{month:02d}" stats_key = f"user:{user_id}:stats:{year}{month:02d}" result = self.redis.bitfield(month_key) .get('u32', 0) .execute() signed_map = result[0] if result else 0 signed_days = bin(signed_map).count('1') stats = self.redis.bitfield(stats_key) .get('u8', 0) # 签到天数 .get('u8', 1) # 最大连续 .get('u8', 2) # 奖励状态 .execute() return { 'signed_map': signed_map, 'total_days': stats[0] if stats else 0, 'max_streak': stats[1] if stats else 0, 'rewards': stats[2] if stats else 0 } def get_current_streak(self, user_id: int) -> int: """获取当前连续签到天数""" result = self.redis.bitfield(f"user:{user_id}:streak") .get('u16', 0) .execute() return result[0] if result else 0

实际部署时，建议配合Lua脚本实现更复杂的原子操作。例如处理跨天签到逻辑：

local function handle_cross_day_sign(user_id) local current_time = redis.call('TIME')[1] local last_sign = redis.call('GET', 'user:'..user_id..':last_sign') -- 如果超过48小时未签到则重置连续天数 if last_sign and (current_time - tonumber(last_sign)) > 172800 then redis.call('BITFIELD', 'user:'..user_id..':streak', 'SET', 'u16', '0', '0') end redis.call('SET', 'user:'..user_id..':last_sign', current_time) return true end