如何用Python实现Redis实时用户数统计？
实时用户数统计是许多在线业务系统的基础能力。社交产品需要展示当前在线人数，电商平台需要按小时统计活跃访客，内容社区需要按天计算去重活跃作者。Redis作为高性能内存数据库，凭借原子操作、丰富的数据结构和低延迟特性，非常适合承载这类高并发计数任务。在Python技术栈中，借助redis-py客户端可以快速实现用户行为采集、去重计数、过期清理等逻辑。本文将围绕当前在线用户数、时间窗口活跃用户数以及异常离线处理三个核心场景，介绍基于Redis的实时统计实现方法。
实际落地时，开发者往往需要先明确统计口径。口径不同，选用的Redis数据结构、计算精度和存储成本也会不同。例如，在线用户数要求精确去重，而活跃用户数在大规模场景下可以接受近似结果。只有理解了统计需求与数据结构特性之间的匹配关系，才能设计出稳定、高效且易于维护的计数方案。
根据统计口径选择合适的数据结构
实时用户数统计并不是单一场景，不同统计口径对数据结构和实现方式有不同要求。常见口径包括：当前在线用户数、某个时间窗口内的活跃用户数、用户最近活跃状态以及是否需要自动过期。当前在线用户数通常要求精确计数，并且需要快速增删成员；活跃用户数则更关注去重后的规模，计数精度可以放宽；而维护用户活跃状态则需要记录最后活跃时间，方便后续清理。
Redis提供了多种数据结构来适配这些场景。SET集合能够存储不重复元素，使用SADD命令添加成员、SREM命令移除成员、SCARD命令获取集合大小，操作复杂度均为O(1)，非常适合精确统计在线用户。HyperLogLog是一种基数估计算法结构，每个键只占用约12KB内存，可以对海量用户ID进行去重计数，标准误差约为0.81%，适合时间窗口活跃用户统计。HASH结构则适合保存用户ID与最后活跃时间戳的映射关系，配合过期时间或定时清理任务，能够处理异常离线场景。
在实际项目中，三种结构可以组合使用。例如，用SET记录在线用户集合，用HASH维护心跳时间戳，用HyperLogLog按天或按小时汇总活跃用户规模。Redis中的命令天然具备原子性，多个服务实例同时操作同一个键时不会产生计数竞争，这也让分布式环境下的实时统计更加可靠。
使用SET结构统计当前在线用户数
当前在线用户数是一个精确计数场景，用户每登录一次，就应该在集合中存在唯一一条记录；用户退出或判定离线后，必须从集合中移除。SET结构天然支持去重，即使用户重复触发登录逻辑，也不会导致重复计数。统计时只需要执行SCARD命令，直接从集合中读取元素数量即可。
下面的代码展示了如何连接Redis、实现用户登录、退出以及获取当前在线人数。在线用户统一存储在键名为online_users的SET中，用户ID作为成员值。为了更好地处理字符串类型，连接时开启decode_responses=True，这样从Redis返回的用户ID就是Python字符串。
import redis
import time
# 连接Redis，默认本机6379端口
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def user_login(user_id):
"""用户登录时加入在线集合"""
r.sadd('online_users', user_id)
print(f"用户{user_id}登录，当前在线用户数：{r.scard('online_users')}")
def user_logout(user_id):
"""用户退出时从在线集合移除"""
r.srem('online_users', user_id)
print(f"用户{user_id}退出，当前在线用户数：{r.scard('online_users')}")
def get_online_count():
"""获取当前在线用户总数"""
return r.scard('online_users')
if __name__ == '__main__':
user_login('user_1001')
user_login('user_1002')
user_login('user_1003')
print(f"当前在线总人数：{get_online_count()}")
time.sleep(2)
user_logout('user_1002')
print(f"当前在线总人数：{get_online_count()}")上述代码中，SADD和SREM都是原子操作，即使多个服务实例同时执行用户登录或退出逻辑，也不会出现集合成员被误删或重复添加的问题。对于在线人数这种相对变化频繁的统计项，SET结构能够在保证精度的同时提供稳定的读写性能。
需要注意的是，纯粹的登录和退出接口无法覆盖用户直接关闭浏览器、断网或进程被杀等异常情况。如果用户没有主动触发退出逻辑，其ID会一直残留在在线集合中，导致统计结果偏高。因此，在线用户统计通常需要配合心跳机制和超时清理策略，这部分内容将在后文专门介绍。
使用HyperLogLog统计时间窗口活跃用户数
活跃用户数通常以某个时间窗口为统计范围，例如最近一小时或最近一天内有过操作的去重用户数量。与当前在线人数不同，活跃用户数不需要精确到每一个用户的实时状态，而是更关注整体规模。当用户量达到百万甚至千万级别时，如果使用SET保存所有用户ID，内存占用会非常可观，而HyperLogLog可以在极低内存开销下完成去重计数。
HyperLogLog的核心思想是通过哈希函数和桶计数来估算基数，Redis提供了PFADD和PFCOUNT命令分别用于添加元素和获取近似基数。重复添加同一用户ID不会增加计数结果，因为内部会对元素进行去重处理。下面的代码使用时间戳整除固定秒数生成递增的时间窗口序号，避免依赖具体日期字符串，同时支持按小时或分钟粒度进行滚动统计。示例代码如下：
import time
import redis
client = redis.Redis()
def record_user_active(user_id, window_seconds=3600):
# 使用当前时间戳整除窗口长度，得到窗口序号
window_id = int(time.time()) // window_seconds
key = f"active:user:window:{window_id}"
# 将用户ID加入 HyperLogLog
client.pfadd(key, user_id)
# 设置过期时间，避免历史窗口永久占用内存
client.expire(key, window_seconds * 2)
def get_active_user_count(window_seconds=3600, offset=0):
window_id = int(time.time()) // window_seconds + offset
key = f"active:user:window:{window_id}"
# 返回近似基数
return client.pfcount(key)上述代码中，window_id将时间轴切分为固定长度的窗口。例如window_seconds=3600时，一小时内的用户行为都会写入同一个 key。设置过期时间为窗口长度的两倍，一方面保证当前窗口内数据可用，另一方面在窗口结束后自动释放陈旧数据。offset参数可以查询上一个完整窗口，便于对比不同时段的活跃趋势。
当需要统计跨多个窗口的活跃用户时，例如最近 3 小时活跃用户，可以调用PFCOUNT同时合并多个 HyperLogLog 键：
# 最近 3 小时窗口
window_seconds = 3600
current = int(time.time()) // window_seconds
keys = [f"active:user:window:{current - i}" for i in range(3)]
approx_count = client.pfcount(*keys)Redis 会合并这些 HyperLogLog 结构并返回近似的总基数。需要注意，HyperLogLog 的计数结果并非精确值，标准误差约为 0.81%。对于活跃用户统计这类业务场景，这个误差通常可以接受。与在线用户统计不同，活跃用户统计往往更关注整体规模而非每个个体的实时状态，因此可以在极低内存消耗下支撑千万级甚至亿级用户规模。
心跳机制与超时清理
前文提到，在线用户统计如果仅依赖登录和退出接口，会存在用户异常退出导致数据残留的问题。解决方案是引入心跳机制：客户端每隔一段时间向服务端发送一次心跳，服务端更新该用户的最后活跃时间，并定期清理超过阈值未更新的用户。
实现上，可以使用有序集合（ZSET）记录每个用户的最后心跳时间，集合（SET）继续用于保存当前在线用户。下面给出一个完整示例：
import time
import redis
client = redis.Redis()
ONLINE_SET = "online:users"
HEARTBEAT_ZSET = "online:heartbeat"
def user_heartbeat(user_id):
now = time.time()
# 将用户加入在线集合
client.sadd(ONLINE_SET, user_id)
# 更新心跳有序集合中的最后活跃时间
client.zadd(HEARTBEAT_ZSET, {user_id: now})
def cleanup_expired_users(timeout_seconds=90):
cutoff = time.time() - timeout_seconds
# 获取所有心跳超时的用户ID
expired_users = client.zrangebyscore(HEARTBEAT_ZSET, '-inf', cutoff)
if not expired_users:
return
# 使用 pipeline 批量删除在线集合和心跳有序集合
pipe = client.pipeline()
pipe.srem(ONLINE_SET, *expired_users)
pipe.zrem(HEARTBEAT_ZSET, *expired_users)
pipe.execute()这里使用 ZSET 存储用户 ID 和最后心跳时间，score 为 Unix 时间戳。清理任务通过ZRANGEBYSCORE获取截止时间之前的所有用户，即超时用户，然后从在线集合和心跳有序集合中批量移除。Pipeline 可以减少多次网络往返，但两个删除操作并非完全原子，在极高并发场景下可能出现短暂不一致。如果要求严格的原子性，可以将清理逻辑封装为 Lua 脚本，在 Redis 服务端串行执行：
-- KEYS[1]: 心跳有序集合 key
-- KEYS[2]: 在线用户集合 key
-- ARGV[1]: 截止时间戳
local expired = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1])
if #expired > 0 then
redis.call('ZREM', KEYS[1], unpack(expired))
redis.call('SREM', KEYS[2], unpack(expired))
end
return #expired执行该脚本时传入对应的 key 和截止时间戳即可：
script = '''
local expired = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1])
if #expired > 0 then
redis.call('ZREM', KEYS[1], unpack(expired))
redis.call('SREM', KEYS[2], unpack(expired))
end
return #expired
'''
cutoff = time.time() - 90
client.eval(script, 2, HEARTBEAT_ZSET, ONLINE_SET, cutoff)心跳间隔与超时阈值需要根据业务场景合理设置。通常心跳间隔可以设置为 30 秒，超时阈值设置为 90 秒或 3 倍心跳间隔，以避免网络抖动导致用户被误清理。清理任务可以由后台定时调度器触发，例如每 30 秒执行一次，从而保持在线用户集合的时效性和准确性。
总结
在线用户统计和活跃用户统计虽然都涉及去重，但对精度、实时性和内存消耗的要求不同。在线用户需要准确反映当前状态，适合使用 SET 并配合心跳机制和超时清理；活跃用户侧重于宏观规模，允许一定误差，HyperLogLog 提供了极低内存占用的高效近似计数方案。根据实际业务选择合适的数据结构，可以在保证统计效果的同时大幅降低系统资源消耗，这也是 Redis 在实际工程中常用的设计思路。