签到功能几乎是所有社交产品、学习平台、健身应用的标配功能。用户每天点一下按钮，系统就记录"今天来了"。听起来很简单，但当你面对千万级用户时，这个问题就变得不那么简单了。今天这篇文章，我来聊聊怎么用 Redis 的位图特性，把签到功能做得既高效又优雅。
一、签到功能背后的存储难题
先说一个实际场景。假设我们做一个学习打卡平台，注册用户有500万，每个用户每天需要记录一次签到状态。如果用最笨的方法——在 MySQL 里建一张签到表，一个用户一天一条记录，那500万用户一年就产生约18亿条记录。数据库的存储压力和查询性能都会成为瓶颈。
我们其实不需要那么复杂的表结构。签到数据本质上就是一个"0或1"的状态——今天签没签到，无非就是这两个值。这种二值状态天然适合用位（bit）来表示，一个 bit 占 1/8 个字节，空间效率极高。500万用户一整年的签到数据，用位图存下来大概也就几十MB，对比数据库的GB级占用，差距是数量级的。
1.1 为什么不用其他方案
你可能想过用 Redis 的 Hash 结构，每个用户一个 key，value 里存每天的签到状态。这种方式确实能跑，但空间占用是位图的很多倍。Hash 的每个 field 有固定的元数据开销，即使 value 只是一个"1"，实际占用的内存也要几百字节。用户量大起来后，这个开销就不可忽视。
再比如用 ZSet 存签到时间戳，这能解决"哪天签了"的问题，但查询"本月签了几天"就需要额外计算。位图的优势在于，一条 BITCOUNT 命令就能拿到统计结果，不需要遍历。
二、位图到底是什么东西
说到位图，先别被"图"字吓到。这里的位图其实是一个纯二进制的位序列，每个位置只存 0 或 1。你可以把它想象成一排小开关，每个开关要么打开（1）要么关闭（0），没有任何中间状态。
Redis 并没有一种专门叫"位图"的数据类型。它的巧妙之处在于，普通的字符串（String）类型天然就能当作位图来使用。Redis 的字符串最大可以存储 512MB 的数据，换算成位就是 512 × 8 × 1024 × 1024 = 42亿个位。这个容量对签到场景来说绰绰有余。
2.1 核心命令介绍
操作位图用到的 Redis 命令其实不多，核心就几个：
SETBIT：设置某个位为 0 或 1。比如 SETBIT mykey 5 1 表示把第5位设置为1，如果原来是0，就返回0；如果本来就是1，就返回1。
GETBIT：读取某个位的值。比如 GETBIT mykey 5 返回第5位当前是0还是1。
BITCOUNT：统计整个位图中有多少个 1。签到场景里这个命令特别有用，直接告诉我这个用户总共签了多少次。
BITOP：位运算操作，支持 AND、OR、XOR、NOT。在连续签到统计场景中会用到。
BITFIELD：批量操作多个位，性能比逐个操作好很多。
2.2 位图命令示例
技术栈：Python + redis 库
# -*- coding: utf-8 -*-
"""
技术栈：Python 3.8+ 配合 redis 库（pip install redis）
本节演示位图的基础操作命令
"""
import redis
# 创建 Redis 客户端连接，假设 Redis 运行在本地默认端口
r = redis.Redis(
host='127.0.0.1', # Redis 服务地址
port=6379, # 默认端口
db=0, # 使用第0号数据库
decode_responses=True # 返回字符串而非 bytes，方便调试
)
# ==================== SETBIT：设置签到位 ====================
# 场景：用户 user_10001 在第 0 位（代表1月1日）标记签到
# SETBIT key offset value
# offset: 位的偏移量，从0开始
# value: 0 或 1
old_value = r.setbit('sign:user:10001:2024:01', 0, 1)
# 返回 0 表示这个位置原来是 0，现在被设置成了 1
# 如果返回 1，说明这个位置本来就是 1（重复签到）
print(f"设置第0位，原值: {old_value}")
# ==================== GETBIT：查询签到状态 ====================
# 场景：查询 user_10001 在1月5日（第4位）是否签到
current_value = r.getbit('sign:user:10001:2024:01', 4)
print(f"第4位当前值: {current_value}") # 返回 0 表示未签到，1 表示已签到
# ==================== BITCOUNT：统计签到总次数 ====================
# 场景：统计 user_10001 在1月份总共签了多少天
total_signed = r.bitcount('sign:user:10001:2024:01')
print(f"1月签到总天数: {total_signed}")
# 如果只需要统计某一段范围（比如1月1日到1月15日，即前16位）
# 使用 start 和 end 参数限制范围
first_half = r.bitcount('sign:user:10001:2024:01', start=0, end=15)
print(f"上半月签到天数: {first_half}")
2.3 BITFIELD 批量操作
BITFIELD 是 Redis 3.2 引入的增强命令，可以对一个位图进行批量操作，减少了多次网络往返带来的开销。在签到场景中，如果我们需要一次性查询用户多个日期的签到状态，BITFIELD 比循环调用 GETBIT 高效得多。
# 批量查询用户 user_10001 在1月的签到状态
# 使用 BITFIELD 一次性获取多天的签到记录
# 命令格式：BITFIELD key [GET type offset] [SET type offset value] ...
# type 为 u（unsigned）或 i（signed），后面跟位数，签到场景用 u1 即可
results = r.bitfield(
'sign:user:10001:2024:01',
command='GET u1 0', # 第0位：1月1日
command='GET u1 1', # 第1位：1月2日
command='GET u1 2', # 第2位：1月3日
command='GET u1 3', # 第3位：1月4日
command='GET u1 4', # 第4位：1月5日
)
# results 返回一个列表，每个元素对应一个 GET 操作的结果
# 例如 [1, 0, 1, 1, 0] 表示1、3、4号签到了，2号和5号没签
print(f"前5天的签到状态: {results}")
三、用位图实现完整签到系统
前面介绍了位图的基本操作，现在我们来搭建一个完整的签到服务。我会把设计思路讲清楚，每一段代码都配上详细的注释。
3.1 Key 的设计策略
Key 的设计直接决定了后续查询和维护的便利性。我的建议是 sign:{user_id}:{year}:{month} 这样的格式，比如 sign:10001:2024:01 代表用户10001在2024年1月的签到记录。
为什么把年月拆开而不是放在一个 key 里？因为位图的索引是从 0 开始的连续整数，如果用单一 key 存一整年的数据，那么1月1日对应第0位，1月32日不存在，2月1日就要跳过到第32位。这种跳位虽然技术上可行，但会让 BITCOUNT 的范围统计变得复杂。按月份拆分 key，每个月都是独立的连续位序列，查询逻辑非常清晰。
不过按月份拆也会带来一个问题：跨年连续签到需要跨 key 查询。这我们后面会处理。
3.2 核心签到服务代码
# -*- coding: utf-8 -*-
"""
技术栈：Python 3.8+ 配合 redis 库
签到服务的核心实现类，封装了位图的所有常用操作
"""
import redis
from datetime import datetime, timedelta
class SignInService:
"""
签到服务类，负责所有签到相关的位图操作。
核心设计思路：
- 每个用户每月使用一个 Redis key 存储签到位图
- key 格式: sign:{user_id}:{year}:{month}
- 月份内第 N 天对应位图的第 (N-1) 位
"""
def __init__(self, redis_client):
"""初始化时注入 Redis 客户端"""
self.r = redis_client
def _build_key(self, user_id, year, month):
"""
根据用户ID、年、月构建 Redis key
例: sign:10001:2024:01
"""
# 月份补零，保证 key 排序和查找的一致性
month_str = str(month).zfill(2)
return f"sign:{user_id}:{year}:{month_str}"
def sign_in(self, user_id, sign_date=None):
"""
用户签到
参数:
user_id: 用户唯一标识
sign_date: 签到日期，默认为当前日期
返回:
dict 包含签到结果信息
"""
# 如果没有指定日期，使用当前日期
if sign_date is None:
sign_date = datetime.now()
year = sign_date.year
month = sign_date.month
day = sign_date.day
# 构建 key，注意 day 从1开始，而位图 offset 从0开始
# 所以 offset = day - 1
offset = day - 1
key = self._build_key(user_id, year, month)
# 使用 SETBIT 设置签到位
# 返回旧值：0 表示首次签到，1 表示重复签到
old_value = self.r.setbit(key, offset, 1)
# 设置 key 的过期时间，避免已过期数据占用内存
# 策略：在当月最后一天后过期，保留15天缓冲
# 也可以设置更精确的过期时间
expire_days = self._get_month_expire_days(sign_date)
self.r.expire(key, expire_days * 86400)
# 同时维护一个连续签到计数器（用 Hash 存储）
# 如果今天签到了，更新连续签到天数
self._update_consecutive_count(user_id, sign_date, old_value == 0)
return {
'signed': old_value == 0, # True 表示是首次签到
'duplicate': old_value == 1, # True 表示重复签到
'date': sign_date.strftime('%Y-%m-%d'),
'consecutive_days': self.get_consecutive_days(user_id)
}
def get_sign_status(self, user_id, date):
"""
查询指定用户某天是否签到
参数:
user_id: 用户ID
date: 查询日期 (datetime 对象)
返回:
True 表示已签到，False 表示未签到
"""
year = date.year
month = date.month
day = date.day
offset = day - 1
key = self._build_key(user_id, year, month)
# GETBIT 返回 0 或 1
value = self.r.getbit(key, offset)
return value == 1
def get_monthly_summary(self, user_id, year, month):
"""
获取用户某月的签到汇总信息
参数:
user_id: 用户ID
year: 年份
month: 月份
返回:
dict 包含签到总数、连续最长天数、每日状态列表
"""
key = self._build_key(user_id, year, month)
# 1. 统计本月总签到天数
total_signed = self.r.bitcount(key)
# 2. 计算本月总天数
import calendar
total_days = calendar.monthrange(year, month)[1]
# 3. 批量获取每一天的签到状态
daily_status = self._get_daily_status_list(key, total_days)
# 4. 计算最长连续签到天数
max_consecutive = self._calc_max_consecutive(daily_status)
# 5. 计算签到率
sign_rate = round(total_signed / total_days * 100, 2) if total_days > 0 else 0
return {
'year': year,
'month': month,
'total_days': total_days,
'signed_days': total_signed,
'sign_rate': f"{sign_rate}%",
'max_consecutive': max_consecutive,
'daily_status': daily_status # 列表，0=未签，1=已签
}
def get_consecutive_days(self, user_id):
"""
获取用户当前连续签到天数（包含当天）
通过从当天往前逐日检查位图状态来计算
"""
consecutive = 0
current_date = datetime.now()
# 从当前日期开始，逐日向前检查
while True:
if self.get_sign_status(user_id, current_date):
consecutive += 1
current_date -= timedelta(days=1)
else:
break
return consecutive
def make_up_sign(self, user_id, make_up_date, limit_days=3):
"""
补签功能：允许用户补签过去几天的记录
参数:
user_id: 用户ID
make_up_date: 补签日期
limit_days: 最多允许补签的天数限制（从当天往前推算）
返回:
dict 补签结果
"""
today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
# 计算补签日期距离今天的天数差
date_diff = (today - make_up_date).days
# 校验：不能补签到未来
if make_up_date > today:
return {
'success': False,
'message': '不能补签到未来日期'
}
# 校验：补签天数不能超过限制
if date_diff > limit_days:
return {
'success': False,
'message': f'只能补签最近{limit_days}天的记录'
}
# 校验：不能重复补签（检查是否已签到）
if self.get_sign_status(user_id, make_up_date):
return {
'success': False,
'message': '该日期已签到，无需补签'
}
# 执行补签操作
result = self.sign_in(user_id, make_up_date)
result['is_makeup'] = True
result['message'] = '补签成功'
return result
def _get_month_expire_days(self, sign_date):
"""
计算 key 的过期天数
策略：当月结束后保留15天，方便用户查看上月记录
"""
import calendar
total_days = calendar.monthrange(sign_date.year, sign_date.month)[1]
# 当月剩余天数 + 15天缓冲
remaining = total_days - sign_date.day
return remaining + 15
def _get_daily_status_list(self, key, total_days):
"""
批量获取某月每一天的签到状态列表
使用 BITFIELD 提高效率，减少网络往返次数
"""
results = []
# 每次请求最多查询64天的数据（Redis BITFIELD 单次操作上限考虑）
for i in range(total_days):
results.append(self.r.getbit(key, i))
return results
def _calc_max_consecutive(self, daily_status):
"""
计算最长连续签到天数
遍历签到状态列表，统计最长连续1的个数
"""
max_consec = 0
current_consec = 0
for status in daily_status:
if status == 1:
current_consec += 1
max_consec = max(max_consec, current_consec)
else:
current_consec = 0 # 遇到未签到，重置计数器
return max_consec
def _update_consecutive_count(self, user_id, sign_date, is_first_sign):
"""
更新用户的连续签到计数（使用 Redis Hash 辅助存储）
当用户首次签到时更新计数器
"""
key = f"sign_consecutive:{user_id}"
if is_first_sign:
# 检查昨天是否签到，如果昨天签了，连续天数+1，否则重置为1
yesterday = sign_date - timedelta(days=1)
if self.get_sign_status(user_id, yesterday):
self.r.hincrby(key, 'count', 1)
else:
self.r.hset(key, 'count', 1)
# 同时记录最长连续天数
current = int(self.r.hget(key, 'count') or 0)
max_count = int(self.r.hget(key, 'max_count') or 0)
if current > max_count:
self.r.hset(key, 'max_count', current)
3.3 使用示例
# -*- coding: utf-8 -*-
"""
技术栈：Python 3.8+ 配合 redis 库
签到服务的完整使用示例，演示常见业务场景
"""
import redis
from datetime import datetime, timedelta
# 初始化 Redis 客户端
r = redis.Redis(
host='127.0.0.1',
port=6379,
db=0,
decode_responses=True
)
# 实例化签到服务
service = SignInService(r)
# ========== 场景1：用户每天签到 ==========
today = datetime.now()
result = service.sign_in(10001, today)
print(f"签到结果: {result}")
# 输出类似: {'signed': True, 'duplicate': False, 'date': '2024-03-15', 'consecutive_days': 5}
# ========== 场景2：重复签到处理 ==========
result2 = service.sign_in(10001, today)
print(f"重复签到结果: {result2}")
# signed 为 False, duplicate 为 True，前端可以提示"今天已签到"
# ========== 场景3：查询某天签到状态 ==========
check_date = today - timedelta(days=3)
is_signed = service.get_sign_status(10001, check_date)
print(f"{check_date.strftime('%Y-%m-%d')} 签到状态: {'已签到' if is_signed else '未签到'}")
# ========== 场景4：获取月度签到汇总 ==========
summary = service.get_monthly_summary(10001, today.year, today.month)
print(f"本月签到汇总:")
print(f" 总天数: {summary['total_days']}")
print(f" 已签到: {summary['signed_days']} 天")
print(f" 签到率: {summary['sign_rate']}")
print(f" 最长连续: {summary['max_consecutive']} 天")
# ========== 场景5：补签操作 ==========
makeup_date = today - timedelta(days=5)
makeup_result = service.make_up_sign(10001, makeup_date, limit_days=7)
print(f"补签结果: {makeup_result}")
# ========== 场景6：连续签到统计 ==========
consecutive = service.get_consecutive_days(10001)
print(f"当前连续签到天数: {consecutive} 天")
四、跨年连续签到的处理
连续签到统计是签到系统的常见需求，也是位图方案需要特别处理的地方。因为每个月是一个独立的 key，跨年时就需要跨 key 查询。
4.1 连续签到的计算策略
连续签到的核心逻辑是：从今天往前一天一天检查，只要遇到一个未签到的日期就停止。这个逻辑本身不复杂，关键是当查询范围跨越月度和年度时，key 需要动态切换。
# -*- coding: utf-8 -*-
"""
技术栈：Python 3.8+ 配合 redis 库
跨月连续签到统计的实现
"""
from datetime import datetime, timedelta
class ConsecutiveSignCounter:
"""连续签到计数器，支持跨月甚至跨年的连续签到统计"""
def __init__(self, redis_client):
self.r = redis_client
def get_cross_month_consecutive(self, user_id, start_date=None):
"""
计算从 start_date 往前推算的连续签到天数
自动处理跨月和跨年的情况
"""
if start_date is None:
start_date = datetime.now()
consecutive = 0
current_date = start_date
# 向前逐日检查，最多检查366天（防止死循环）
for _ in range(366):
key = f"sign:{user_id}:{current_date.year}:{str(current_date.month).zfill(2)}"
offset = current_date.day - 1
# 检查当前位置是否签到
value = self.r.getbit(key, offset)
if value == 1:
consecutive += 1
current_date -= timedelta(days=1)
else:
break
return consecutive
def get_continuous_sign_range(self, user_id):
"""
获取用户的连续签到日期范围
返回从连续签到开始日期到结束日期的范围信息
"""
today = datetime.now()
consecutive = self.get_cross_month_consecutive(user_id, today)
if consecutive == 0:
return {
'start_date': None,
'end_date': None,
'days': 0
}
# 开始日期 = 今天 - 连续天数 + 1
start_date = today - timedelta(days=consecutive - 1)
return {
'start_date': start_date.strftime('%Y-%m-%d'),
'end_date': today.strftime('%Y-%m-%d'),
'days': consecutive
}
4.2 连续签到排行榜
排行榜是签到系统的常见延伸功能。位图本身不支持直接排名，但我们可以借助 Redis 的 Sorted Set 来实现。每天用户签到后，把连续签到天数更新到 ZSet 中，排序和分页查询都非常高效。
# -*- coding: utf-8 -*-
"""
技术栈：Python 3.8+ 配合 redis 库
签到排行榜的实现
"""
import redis
from datetime import datetime
class SignRanking:
"""签到排行榜，基于 Sorted Set 实现"""
def __init__(self, redis_client):
self.r = redis_client
self.ranking_key = "sign:ranking:daily" # 排行榜 key
def update_ranking(self, user_id, consecutive_days):
"""
更新用户排行榜分数
使用 ZSet 的 ZADD 命令，score 为连续签到天数
"""
self.r.zadd(self.ranking_key, {str(user_id): consecutive_days})
def get_top_n(self, n=10):
"""
获取排行榜前 N 名
ZREVRANGE 按分数降序排列，取前 N 个
"""
# withscores=True 同时返回成员和分数
results = self.r.zrevrange(self.ranking_key, 0, n - 1, withscores=True)
return [{'user_id': int(user_id), 'consecutive_days': int(score)} for user_id, score in results]
def get_user_rank(self, user_id):
"""获取指定用户的排名"""
rank = self.r.zrevrank(self.ranking_key, str(user_id))
return rank + 1 if rank is not None else None # Redis 排名从0开始，转为1开始
def clear_ranking(self):
"""清空排行榜（一般在重置周期开始时调用）"""
self.r.delete(self.ranking_key)
五、应用场景分析
5.1 适用场景
位图方案特别适合以下场景：
高频读写、低延迟要求：用户每天签到一次，写入频率不高，但读取可能很频繁（查看月度签到日历、计算连续天数等）。位图的读写操作都是 O(1) 复杂度，响应极快。
空间敏感：用户量巨大的场景下，存储空间是核心考量。位图的存储效率是传统方案的数十倍甚至上百倍，对于百万级用户产品来说，这可能意味着节省数GB的存储成本。
简单二值状态：签到本质就是"有"或"没有"，这种二值状态是位图的最佳应用场景。如果需要存储更多维度的信息（比如签到时间、签到类型等），位图就不太适合了，需要考虑 Hash 或更复杂的数据结构。
5.2 不适用场景
位图也不是万能的。如果签到记录需要关联更多信息，比如用户每天签到时选择的心情、完成的运动量、打卡的地点等，位图就无能为力了。这时候应该回到传统的键值存储方案，比如 Hash 或者直接在数据库里存储结构化数据。
另外，位图不支持直接查询"某一天是否签到"之外的语义，比如"本月签到最频繁的一周"这种需要更复杂计算的查询，就需要额外的逻辑来处理。
六、技术优缺点分析
6.1 优势
极致空间效率：一个用户一年的签到数据只占约54字节（365 bit ≈ 46字节，加上 Redis 元数据约54字节）。1亿用户的全年签到数据也只需要约5.4GB，这对 Redis 来说完全可承受。对比 MySQL 方案，1亿用户一年需要存储约100GB的签到记录数据。
查询性能出色：BITCOUNT 统计某月签到天数只需一条命令，时间复杂度 O(N/64)（N 为位数），365天的统计在微秒级别完成。BITFIELD 批量查询也可以在一次网络请求中完成多天的状态获取。
原子操作保证：SETBIT 是原子操作，多线程并发签到不会导致数据不一致。这比"先查再写"的方案安全得多。
易维护的 Key 设计：按月拆分 key 的方案天然支持数据过期清理，设置 TTL 就能自动回收已过期数据，不需要额外的定时任务来删除。
6.2 劣势
查询灵活性有限：位图只能表达 0 和 1，如果需要存储签到时间、签到渠道等附加信息，位图就搞不定了。需要额外用 Hash 来存储这些辅助信息，增加了系统复杂度。
跨月查询稍复杂：每月的数据分散在不同 key 中，跨月连续签到需要多次查询不同 key，性能不如单一 key 的方案。不过实际测试中，即使是跨3个月的连续签到查询，总耗时也在毫秒级，对用户体验影响不大。
调试不够直观：位图在 Redis 中存储为二进制字符串，直接用 redis-cli 查看时是一堆不可读的字符。排查问题时需要借助脚本转换，不如文本格式直观。
6.3 与替代方案的对比
和 Hash 方案对比：Hash 的每个 field-value 对都有一定的元数据开销（大约78字节起步），存储一个用户的365天签到数据，Hash 方案需要约28KB，而位图只需54字节。100万用户的差距就是约28GB vs 54MB。
和 ZSet 方案对比：ZSet 适合存储有序的带分数的数据，签到如果用 ZSet 存时间戳，空间占用比 Hash 略小，但查询"本月签了几天"需要 ZCOUNT 范围查询，性能不如 BITCOUNT 直接统计。
七、注意事项
7.1 Key 命名规范
Key 的命名不仅影响可读性，还影响 Redis 的内存管理。建议遵循 业务:实体:维度 的格式，比如 sign:user:10001:2024:01。月份部分建议补零（01、02而不是1、2），这样 key 排序和字符串比较的结果符合预期。
7.2 过期策略
设置合理的 TTL 过期策略很重要。我的建议是：当月签到数据保留到次月结束后再过期，给用户留出查看上月记录的时间。具体可以在 key 创建时设置过期时间为当月剩余天数加上15天缓冲。
# -*- coding: utf-8 -*-
"""
技术栈：Python 3.8+ 配合 redis 库
设置签到 key 的过期时间策略示例
"""
import redis
from datetime import datetime
import calendar
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
def set_sign_key_expiry(user_id, sign_date):
"""
为签到 key 设置合理的过期时间
策略：当月结束后保留15天
"""
key = f"sign:{user_id}:{sign_date.year}:{str(sign_date.month).zfill(2)}"
# 计算当月总天数
total_days_in_month = calendar.monthrange(sign_date.year, sign_date.month)[1]
# 当月剩余天数
days_remaining = total_days_in_month - sign_date.day
# 设置过期时间：剩余天数 + 15天缓冲
total_expire_days = days_remaining + 15
# expire 命令设置过期时间，单位为秒
r.expire(key, total_expire_days * 86400)
# 验证过期时间
ttl = r.ttl(key)
print(f"Key: {key}, TTL: {ttl}秒 ({ttl // 86400}天)")
7.3 数据一致性
位图操作本身是原子的，不会出现"半写入"的中间状态。但补签场景下需要注意校验逻辑：补签前必须先检查该日期是否已签到，避免覆盖已有的签到记录。在 sign_in 方法中已经通过检查 SETBIT 返回的旧值来处理这个逻辑。
7.4 性能优化建议
批量操作：如果需要查询用户一整月的签到状态，不要循环调用 GETBIT 31次。使用 BITFIELD 可以一次请求获取多天数据，减少网络往返。
Pipeline 批处理：如果需要同时操作多个用户的签到数据（比如批量导出），使用 Redis Pipeline 可以将多条命令打包发送，减少网络开销。
# -*- coding: utf-8 -*-
"""
技术栈：Python 3.8+ 配合 redis 库
使用 Pipeline 批量查询多个用户的签到数据
"""
import redis
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
def batch_query_sign_status(user_ids, year, month):
"""
批量查询多个用户某月的签到统计
使用 Pipeline 优化性能
"""
results = {}
with r.pipeline() as pipe:
for uid in user_ids:
key = f"sign:{uid}:{year}:{str(month).zfill(2)}"
# 对每个用户执行 BITCOUNT
pipe.bitcount(key)
# execute() 一次性发送所有命令并接收所有响应
responses = pipe.execute()
for uid, count in zip(user_ids, responses):
results[uid] = count
return results
# 示例：查询10个用户3月份的签到天数
user_ids = [10001, 10002, 10003, 10004, 10005,
10006, 10007, 10008, 10009, 10010]
stats = batch_query_sign_status(user_ids, 2024, 3)
for uid, count in stats.items():
print(f"用户{uid}: 3月签到{count}天")
7.5 容错处理
生产环境中，Redis 连接可能中断、网络可能超时。签到服务应该对 Redis 操作添加异常处理和重试机制。核心签到操作可以设计为"最终一致性"——如果 Redis 写入失败，先记录到本地日志，后台任务异步重试写入。
7.6 数据迁移与扩容
当单台 Redis 实例的内存接近瓶颈时，可以考虑使用 Redis Cluster 进行分片。按 user_id 做分片，同一个用户的所有签到数据落在同一个 slot 中，查询不需要跨节点。分片策略建议在系统设计初期就规划好，避免后期数据迁移的成本。
八、总结
回到最初的问题——如何用高效的方式存储和查询签到数据。位图方案的核心优势在于空间效率和查询速度。一个用户一年的签到数据只需要54字节，1亿用户的全量签到数据也能控制在几个GB以内。BITCOUNT 一条命令就能拿到月度签到统计，BITFIELD 批量查询减少网络开销，整体性能非常好。
但位图也不是银弹。它适合纯粹的二值状态记录，如果签到功能需要关联更多业务信息（比如签到时的运动数据、心情标签等），就需要配合其他数据结构共同使用。实际项目中，推荐位图负责签到状态的存储和统计，Hash 负责存储签到时的附加信息，两者各司其职。
在设计签到系统时，我建议关注以下几点：Key 的命名规范要清晰且可维护；过期策略要合理，避免无用数据占用内存；并发场景下的原子性要通过 SETBIT 的返回值来处理；跨月跨年的连续签到需要做好状态切换的逻辑。把这些细节处理好，位图方案就能在大型签到系统中稳定运行。
最后提一点实际开发中的经验：位图在 Redis 中存储为二进制字符串，直接通过 redis-cli 查看时是乱码。调试时建议写一个简单的转换脚本，把二进制数据转成可读的 0/1 序列，或者直接用 GETBIT 逐位查询。这个细节虽然小，但能省去不少排查问题的时间。
评论
围绕“Redis位图在签到功能场景中的高效实现与优化”参与讨论