Redis 从入门到精通：位图、HyperLogLog、GEO

IT策士 10余年一线大厂经验，专注 IT 思维、架构、职场进阶。我会在各个平台持续发布最新文章，助你少走弯路。

前四篇，我们把 Redis 五大基础数据结构（String、Hash、List、Set、Sorted Set）全部吃透了。你已经能用它们解决缓存、计数、对象存储、消息队列、排行榜等绝大多数业务问题。

但 Redis 的“兵器库”里还有三件常被忽视的“特种兵器”：位图（Bitmap）、HyperLogLog和GEO。它们不是独立的第六、七、八种数据类型——本质上，位图和 HyperLogLog 是 String 的特殊用法，GEO 是 Sorted Set 的一层封装。但它们解决的问题太特殊、太好用了，值得单独一篇文章来讲述。

• 想用极低内存记录海量用户的签到状态？→ 位图

• 想统计亿级 UV而内存只占 12KB？→ HyperLogLog

• 想实现“附近的人”功能且支持距离排序？→ GEO

读完这篇，你的 Redis 工具箱将再无死角。

1. 位图（Bitmap）—— 海量二值状态的极致压缩

1.1 位图是什么？

位图不是新类型，它就是String。只不过我们不再把 String 看成字符串，而是看成一个比特数组，每个比特位（0 或 1）代表一个布尔状态。

String 值:"abc"(3字节=24比特)比特位:011000010110001001100011偏移量:012345678910...

Redis 提供了直接操作比特位的命令，我们可以把任意 offset 位置的值设为 0 或 1。这样，用 1 个比特就能记录一个用户的签到、一个设备的在线状态，内存利用率是 String 直接存用户 ID 的上万倍。

💡 按理论值：1MB 的位图可以记录 800 多万个用户的状态。

1.2 核心命令速览

# 设置偏移量 offset 的比特值SETBIT key offset value# value 只能是 0 或 1# 获取偏移量 offset 的比特值GETBIT key offset# 统计值为 1 的比特位个数BITCOUNT key[start end]# start/end 是字节位置，不是比特！# 位图间的位运算（AND, OR, XOR, NOT），结果存入新 keyBITOP operation destkey key1[key2...]# 查找第一个值为 bit 的位置BITPOS key bit[start][end]

动手体验：

# 模拟用户签到，用户 ID 作为 offset127.0.0.1:6379>SETBIT sign:2026-06-1010011(integer)0127.0.0.1:6379>SETBIT sign:2026-06-1010051(integer)0127.0.0.1:6379>SETBIT sign:2026-06-1020001(integer)0# 查询用户是否签到127.0.0.1:6379>GETBIT sign:2026-06-101001(integer)1127.0.0.1:6379>GETBIT sign:2026-06-101002(integer)0# 统计签到总人数127.0.0.1:6379>BITCOUNT sign:2026-06-10(integer)3# 查找第一个签到用户的 offset127.0.0.1:6379>BITPOS sign:2026-06-101(integer)1001

1.3 Python 实战：用户签到系统

场景：记录用户每日签到、查询连续签到天数（简化版），以及按月统计活跃天数。

importredis from datetimeimportdate, timedelta r=redis.Redis(host='localhost',port=6379,decode_responses=True)class DailySignIn:"""基于位图的用户签到系统""" def __init__(self,prefix='sign'): self.prefix=prefix def _key(self, dt):"""生成键名，如 sign:2026-06-10"""returnf'{self.prefix}:{dt.strftime("%Y-%m-%d")}'def sign(self, user_id,dt=None):"""用户签到"""ifdt is None: dt=date.today()returnr.setbit(self._key(dt), user_id,1)def is_signed(self, user_id,dt=None):"""检查是否签到"""ifdt is None: dt=date.today()returnr.getbit(self._key(dt), user_id)==1def count_signed(self,dt=None):"""统计当天签到人数"""ifdt is None: dt=date.today()returnr.bitcount(self._key(dt))def continuous_days(self, user_id,end_date=None,max_days=30):"""计算从 end_date 往前连续签到天数（简化版）"""ifend_date is None: end_date=date.today()# 用 BITFIELD 批量获取多个比特位（也可以逐天 GETBIT）days=0foriinrange(max_days): check_date=end_date - timedelta(days=i)ifr.getbit(self._key(check_date), user_id)==1: days+=1else:breakreturndays def monthly_active_mask(self, user_id, year, month):"""生成某月每天签到情况的二进制字符串(1=已签,0=未签)"""importcalendar _, days_in_month=calendar.monthrange(year, month)mask=''fordayinrange(1, days_in_month +1): dt=date(year, month, day)mask+='1'ifself.is_signed(user_id, dt)else'0'returnmask# ----------------- 测试 -----------------sign_in=DailySignIn()# 模拟用户签到print("用户 1001 签到:", sign_in.sign(1001))# 0 (首次设置返回旧值 0)print("用户 1002 签到:", sign_in.sign(1002))print("用户 1003 签到:", sign_in.sign(1003))# 重复签到（不会产生副作用）print("用户 1001 再次签到:", sign_in.sign(1001))# 1 (已为 1，返回旧值 1)# 检查签到状态print("用户 1001 是否签到:", sign_in.is_signed(1001))# Trueprint("用户 9999 是否签到:", sign_in.is_signed(9999))# False# 统计签到人数print(f"今日签到人数: {sign_in.count_signed()}")# 模拟连续签到天数（需要在之前几天也有数据）# 手动设置过去 3 天的签到foriinrange(1,4): past_date=date.today()- timedelta(days=i)r.setbit(f'sign:{past_date.strftime("%Y-%m-%d")}',1001,1)print(f"用户 1001 连续签到天数: {sign_in.continuous_days(1001)}")

输出示例：

用户1001签到:0用户1002签到:0用户1003签到:0用户1001再次签到:1用户1001是否签到: True 用户9999是否签到: False 今日签到人数:3用户1001连续签到天数:4

🧪动手试试：运行代码后，尝试模拟 1000 个用户签到，用BITCOUNT验证签到人数。然后对比：如果用 String 存储每个用户签到状态（如SET sign:user:1001 1），会占用多少内存？位图方式只占用一个 offset，高下立判。

1.4 位图运算：BITOP

BITOP可以对多个位图执行 AND、OR、XOR、NOT 操作，结果存入一个新 key。这在统计分析中非常强大。

# 两个日期的位图做 AND，得到两天都签到的用户127.0.0.1:6379>SETBIT day111127.0.0.1:6379>SETBIT day121127.0.0.1:6379>SETBIT day211127.0.0.1:6379>SETBIT day231127.0.0.1:6379>BITOP AND both_days day1 day2(integer)1127.0.0.1:6379>BITCOUNT both_days(integer)1# 只有用户 1 两天都签到

Python 示例——统计连续 7 天都签到的用户：

def continuous_7days_users(r, prefix, start_date):"""找出从 start_date 开始连续7天都签到的用户数（简化：用 AND 运算）""" keys=[]foriinrange(7): dt=start_date + timedelta(days=i)keys.append(f'{prefix}:{dt.strftime("%Y-%m-%d")}')# BITOP AND 将所有天的位图做与运算r.bitop('AND','tmp:continuous_7days', *keys)count=r.bitcount('tmp:continuous_7days')r.delete('tmp:continuous_7days')# 清理临时 keyreturncount

2. HyperLogLog —— 海量数据的基数统计

2.1 为什么要 HyperLogLog？

“统计网页 UV” 是一个经典需求。最直观的做法：用 Set 存储每个用户的 ID，SCARD获取总数。但如果日活百万甚至千万，Set 内存开销巨大（一个用户 ID 假设 20 字节，1000 万用户就是 200MB）。

HyperLogLog（简称 HLL）专为此而生：它用极小的固定内存（最多 12KB），以 0.81% 的标准误差估算集合的基数（不重复元素个数）。

⚠️ 注意：HLL 不能获取具体元素，只能估算“不重复的个数”。

2.2 核心命令

# 向 HyperLogLog 添加元素PFADD key element[element...]# 估算基数（不重复元素个数）PFCOUNT key[key...]# 合并多个 HyperLogLog 到新 keyPFMERGE destkey sourcekey[sourcekey...]

命令都很简单，前缀PF是为了纪念发明人 Philippe Flajolet。

127.0.0.1:6379>PFADD uv:page1 user1 user2 user3 user4 user5(integer)1127.0.0.1:6379>PFADD uv:page1 user3 user6# user3 重复，忽略(integer)1127.0.0.1:6379>PFCOUNT uv:page1(integer)6

2.3 Python 实战：网站 UV 统计

importredisimportrandom r=redis.Redis(host='localhost',port=6379,decode_responses=True)class UVTracker:"""基于 HyperLogLog 的 UV 统计器""" def __init__(self): pass def record_visit(self, page, user_id):"""记录一次页面访问"""returnr.pfadd(f'uv:{page}', user_id)def get_uv(self, page):"""获取页面 UV（估算值）"""returnr.pfcount(f'uv:{page}')def merge_uv(self, dest_page, *source_pages):"""合并多个页面的 UV，得到总的独立访客数""" keys=[f'uv:{p}'forpinsource_pages]r.pfmerge(f'uv:{dest_page}', *keys)returnr.pfcount(f'uv:{dest_page}')# ----------------- 测试 -----------------tracker=UVTracker()# 模拟 10000 次页面访问，其中独立用户约 5000 个print("模拟用户访问中...")foriinrange(10000): user_id=f'user_{random.randint(1, 5000)}'# 5000 个独立用户tracker.record_visit('home', user_id)uv=tracker.get_uv('home')print(f"页面 home 的 UV 估算值: {uv}")print(f"实际独立用户数: 5000")print(f"误差: {abs(uv - 5000) / 5000 * 100:.2f}%")# 合并多个页面foriinrange(5000): user_id=f'user_{random.randint(1, 3000)}'tracker.record_visit('product', user_id)total=tracker.merge_uv('total','home','product')print(f"合并 home + product 去重 UV 估算: {total}")print(f"实际去重 UV (理论上限): 5000 + 3000 = 8000，但会有重叠")# 验证内存占用print(f"\n内存占用（约）: {r.memory_usage('uv:home')} bytes")

输出示例：

模拟用户访问中... 页面 home 的 UV 估算值:4987实际独立用户数:5000误差:0.26% 合并 home + product 去重 UV 估算:6459内存占用（约）:12304bytes

🧪动手试试：尝试修改代码，把user_id的范围改成 1 到 10 万（10 万独立用户），观察 HLL 估算误差是否仍在 1% 以内。

2.4 适用场景与局限

✅适用：

• 网页 / 应用 UV 统计

• 搜索热词去重计数

• 广告曝光去重人数

❌不适用：

• 需要获取具体元素列表

• 需要精确计数（如金钱相关）

• 小数据量（几千以内），直接用 Set 更精确

3. GEO —— 地理位置与“附近的人”

3.1 GEO 是什么？

GEO 是 Redis 3.2+ 引入的地理位置模块，底层基于 Sorted Set。它用 Geohash 算法将经纬度编码为一个数值作为 score，存储在 ZSet 中，从而实现地理位置的距离计算、范围查询等功能。

Redis GEO 命令内部操作的都是 ZSet，所以你也可以用 ZSet 的命令操作 GEO 键。

3.2 核心命令速览

# 添加地理位置GEOADD key longitude latitude member[longitude latitude member...]# 获取位置坐标GEOPOS key member[member...]# 计算两个位置之间的距离GEODIST key member1 member2[m|km|ft|mi]# 获取指定位置的 Geohash 字符串GEOHASH key member[member...]# 查询指定半径内的成员GEOSEARCH key FROMMEMBER member BYRADIUS radius unit[WITHCOORD][WITHDIST][COUNT count]GEOSEARCH key FROMLONLAT lon lat BYRADIUS radius unit# GEOSEARCH 是 Redis 6.2+ 统一命令，老版本用 GEORADIUS / GEORADIUSBYMEMBER

快速体验：

# 添加几个地点的经纬度127.0.0.1:6379>GEOADD cities116.39739.908"北京"121.47331.230"上海"113.26423.129"广州"114.05722.543"深圳"(integer)4# 获取坐标127.0.0.1:6379>GEOPOS cities 北京 上海1)1)"116.39700299501419067"2)"39.90799927282371714"2)1)"121.47300094366073608"2)"31.22999903983650159"# 计算距离127.0.0.1:6379>GEODIST cities 北京 上海 km"1066.4336"# 查询深圳附近 200km 内的城市# 使用 GEOSEARCH (Redis 6.2+)127.0.0.1:6379>GEOSEARCH cities FROMMEMBER 深圳 BYRADIUS200km WITHDIST1)1)"深圳"2)"0.0000"2)1)"广州"2)"104.7797"# 如果 Redis < 6.2，用 GEORADIUSBYMEMBER127.0.0.1:6379>GEORADIUSBYMEMBER cities 深圳200km WITHDIST1)1)"深圳"2)"0.0000"2)1)"广州"2)"104.7797"

3.3 Python 实战：“附近的人”功能

场景：用户打开 App，按距离排序查询附近的商家。

importredisimportrandomimportmath r=redis.Redis(host='localhost',port=6379,decode_responses=True)class NearbyService:"""基于 GEO 的“附近的人/商家”服务""" GEO_KEY='locations:merchants'def add_merchant(self, name, longitude, latitude):"""添加商家位置"""returnr.geoadd(self.GEO_KEY,(longitude, latitude, name))def get_position(self, name):"""获取商家坐标""" pos=r.geopos(self.GEO_KEY, name)ifpos and pos[0]:return{'longitude':pos[0][0],'latitude':pos[0][1]}returnNone def get_distance(self, name1, name2,unit='km'):"""获取两个商家之间的距离"""returnr.geodist(self.GEO_KEY, name1, name2,unit=unit)def search_nearby(self, longitude, latitude, radius,unit='km',count=None):"""查询指定坐标附近 radius 范围内的商家，按距离排序""" kwargs={'radius':radius,'unit':unit,'withdist':True,'withcoord':True,'sort':'ASC'}ifcount: kwargs['count']=count# GEOSEARCH 在 redis-py 4.0+ 可用results=r.geosearch(self.GEO_KEY,longitude=longitude,latitude=latitude, **kwargs)merchants=[]foriteminresults: merchants.append({'name':item[0],'distance':item[1],'longitude':item[2][0],'latitude':item[2][1]})returnmerchants def search_nearby_member(self, member, radius,unit='km',count=None):"""查询指定商家附近的其他商家""" kwargs={'radius':radius,'unit':unit,'withdist':True,'withcoord':True,'sort':'ASC'}ifcount: kwargs['count']=count results=r.geosearch(self.GEO_KEY,member=member, **kwargs)merchants=[]foriteminresults: merchants.append({'name':item[0],'distance':item[1],'longitude':item[2][0],'latitude':item[2][1]})returnmerchants# ----------------- 测试 -----------------service=NearbyService()# 初始化一批北京和上海周边的商家merchants=[# 北京周边('星巴克(国贸店)',116.461,39.909),('海底捞(三里屯)',116.455,39.932),('麦当劳(望京店)',116.491,39.998),('全聚德(前门店)',116.397,39.899),# 上海周边('星巴克(南京路店)',121.475,31.233),('海底捞(陆家嘴店)',121.508,31.237),('喜茶(人民广场)',121.473,31.230),('外婆家(静安寺)',121.446,31.224),]forname, lon, latinmerchants: service.add_merchant(name, lon, lat)print("=== 查询“我”附近的商家 ===")# 假设我在北京国贸附近my_lon, my_lat=116.460,39.905nearby=service.search_nearby(my_lon, my_lat,radius=5,unit='km')forminnearby: print(f" {m['name']} — {m['distance']:.2f} km")print("\n=== 查询“星巴克(国贸店)”5km 内的其他商家 ===")nearby_starbucks=service.search_nearby_member('星巴克(国贸店)',radius=5,unit='km')forminnearby_starbucks: print(f" {m['name']} — {m['distance']:.2f} km")print("\n=== 北京上海两地距离 ===")dist=service.get_distance('全聚德(前门店)','喜茶(人民广场)',unit='km')print(f"全聚德(前门店) ↔ 喜茶(人民广场): {dist} km")

输出示例：

===查询“我”附近的商家===星巴克(国贸店)—0.46km 海底捞(三里屯)—3.03km 全聚德(前门店)—5.00km===查询“星巴克(国贸店)”5km 内的其他商家===星巴克(国贸店)—0.00km 海底捞(三里屯)—3.49km 全聚德(前门店)—5.00km===北京上海两地距离===全聚德(前门店)↔ 喜茶(人民广场):1066.47km

3.4 GEO 底层与注意事项

GEO 底层是一个 ZSet，score 是坐标的 Geohash 编码（52 位整数）。这意味着：

• 可以用ZREM删除位置、ZCARD获取数量。

• 不要混用 ZADD 直接添加，会破坏 Geohash 编码，导致 GEO 命令异常。

Redis 7 还引入了GEOSEARCHSTORE，可以将查询结果存储为新的 ZSet，方便做二次处理。

4. 总结对比

三大高级结构，各自解决了特定领域的问题，共同特点是：用最小的内存，做最大的事。它们是 Redis 在“基础数据结构”之外的精华扩展，也是你从“会用 Redis”迈向“精通 Redis”的必经之路。

5. 动手试试

在你的 Redis 环境中完成以下挑战：

1. 连续签到统计：模拟 10 个用户过去 7 天的签到数据（用位图），用BITOP AND找出连续 7 天全勤的用户。

2. UV 准确度验证：用同一个数据集（10000 条数据，5000 个唯一用户），同时写入 Set 和 HyperLogLog，对比统计结果，计算误差。

3. 外卖配送范围：录入 20 个商家坐标，模拟用户在某个位置，查询 3km 范围内的商家，按距离排序。

预期效果：1. 全勤用户精准筛选；2. HLL 误差在 1% 以内；3. 附近商家列表距离正确。

从下一篇开始，我们将切换视角，聚焦Python 如何优雅地操作 Redis，深入redis-py的连接池、Pipeline 等工程化技巧，让代码更具生产级风范。

想了解更多还可以去各个平台搜索「IT策士」，一起升级 IT 思维 ！