Redis Bitmaps
以下是对 Redis Bitmaps 的详细介绍与分析，结合其核心特性、应用场景及优化策略：
一、Bitmaps 核心概念
1. 定义与底层原理
-
Bitmaps 是 Redis 基于字符串（String）实现的一种 二进制位数组 结构，每个位（bit）仅存储
0 或1，代表二元状态（如存在/不存在、是/否）。 -
底层实现：使用 SDS（动态字符串）存储二进制数据，通过偏移量（offset）定位位值，计算公式为：
- 字节位置：
offset / 8 - 位位置：
7 - (offset % 8)。
- 字节位置：
2. 核心命令
| 命令 | 功能描述 |
|---|---|
SETBIT key offset 1/0 | 设置偏移量为1或0（如记录用户登录状态）。 |
GETBIT key offset | 获取指定偏移量的位值。 |
BITCOUNT key | 统计值为1的位总数（如计算活跃用户数）。 |
BITOP AND/OR/XOR destkey keys... | 对多个 Bitmaps 执行位运算（交集、并集等）。 |
BITPOS key 1/0 | 查找首个1或0的偏移量（如首次签到时间）。 |
二、应用场景分析
1. 典型场景
| 场景 | 实现方式 | 优势 |
|---|---|---|
| 用户签到 | 每日一个 Bitmap，用户 ID 为偏移量，1表示签到。 | 月度签到仅需 31 bits（约 4B），极大节省空间。 |
| 在线状态统计 | 每个用户 ID 对应一个位，实时更新在线状态。 | 快速查询用户是否在线（O(1)时间复杂度）。 |
| 权限管理 | 每位代表一种权限，1表示拥有（如 Linux 文件权限模型）。 | 通过位运算快速组合权限（如BITOP OR合并权限组）。 |
| 行为分析 | 记录用户点击、浏览等行为，通过BITCOUNT统计高频行为。 | 支持大规模数据实时分析（如千万级用户行为轨迹）。 |
2. 性能对比
- 空间效率：存储
1 亿用户状态仅需 12.5 MB（1e8 / 8）内存，而集合（Set）需数百 MB。 - 时间效率：
SETBIT/GETBIT 为O(1)，BITCOUNT 通过优化算法实现接近O(1)。
三、优缺点与优化策略
1. 优势
- 高存储密度：每个位仅占 0.125 字节，适合海量二元状态存储。
- 原子操作：命令天然支持原子性，避免并发问题。
- 位运算能力：支持复杂逻辑运算（如统计多日活跃用户）。
2. 局限性
- 仅二值状态：无法存储多值信息（需结合其他数据结构）。
- 稀疏数据低效：若数据稀疏（如仅有少数位为
1），集合（Set）可能更省内存。 - 集群限制：Redis Cluster 中所有操作需保证键在同一槽（需用
{tag} 强制分布）。
3. 优化建议
- 偏移量设计：将用户 ID 映射为连续整数，避免哈希散列导致的稀疏偏移。
- 预分配内存：通过提前设置大偏移量（如
SETBIT large-offset 0）避免动态扩容开销。 - 避免大键操作：
BITOP 处理超大 Bitmaps 可能阻塞服务，建议在从节点执行。
四、实际案例：用户签到系统
1. 记录签到
# 用户 1001 在 2025-04-24 签到（偏移量从 0 开始）
SETBIT sign:1001:202504 23 1
2. 统计月度签到次数
BITCOUNT sign:1001:202504
3. 查询首次签到日期
BITPOS sign:1001:202504 1 # 返回偏移量 +1 即为日期
案例代码：
package com.example.redis.bitmaps;import redis.clients.jedis.Jedis;import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;/*** 用户签到服务类* 使用 Redis Bitmap 实现用户签到功能* 键格式：user:sign:{userId}:{year}:{month}*/
public class SignInService {private final Jedis jedis;public SignInService(String host, int port) {this.jedis = new Jedis(host, port);}/*** 用户签到** @param userId 用户ID* @return 是否签到成功*/public boolean signIn(String userId) {LocalDate today = LocalDate.now();// 当月keyString key = generateKey(userId, today);int dayOfMonth = today.getDayOfMonth() - 1; // Redis bitmap 从0开始// 检查是否已经签到if (isSignedIn(userId, today)) {return false;}// 设置签到标记jedis.setbit(key, dayOfMonth, true);return true;}/*** 检查指定日期是否已签到** @param userId 用户ID* @param date 日期* @return 是否已签到*/public boolean isSignedIn(String userId, LocalDate date) {String key = generateKey(userId, date);int dayOfMonth = date.getDayOfMonth() - 1;return jedis.getbit(key, dayOfMonth);}/*** 获取用户当月签到次数** @param userId 用户ID* @return 签到次数*/public long getMonthlySignInCount(String userId) {LocalDate date = LocalDate.now();String key = generateKey(userId, date);return jedis.bitcount(key);}/*** 打印用户当月签到情况** @param userId 用户ID* @return 签到情况列表*/public List<String> getMonthlySignInStatus(String userId) {LocalDate date = LocalDate.now();String key = generateKey(userId, date);int daysInMonth = date.lengthOfMonth();List<String> status = new ArrayList<>();for (int i = 0; i < daysInMonth; i++) {boolean signed = jedis.getbit(key, i);LocalDate currentDate = date.withDayOfMonth(i + 1);String dateStr = currentDate.format(DateTimeFormatter.ISO_DATE);status.add(String.format("%s: %s", dateStr, signed ? "已签到" : "未签到"));}return status;}/*** 获取用户连续签到天数** @param userId 用户ID* @return 连续签到天数*/public int getContinuousSignInDays(String userId) {LocalDate today = LocalDate.now();String key = generateKey(userId, today);int dayOfMonth = today.getDayOfMonth() - 1;int continuousDays = 0;for (int i = dayOfMonth; i >= 0; i--) {if (!jedis.getbit(key, i)) {break;}continuousDays++;}return continuousDays;}/*** 生成 Redis key*/private String generateKey(String userId, LocalDate date) {return String.format("user:sign:%s:%d:%d",userId,date.getYear(),date.getMonthValue());}/*** 关闭 Redis 连接*/public void close() {jedis.close();}public static void main(String[] args) {SignInService signInService = new SignInService("localhost", 6379);try {// 用户签到signInService.signIn("1001");// 检查今日是否已签到boolean signed = signInService.isSignedIn("1001", LocalDate.now());// 获取当月签到统计long count = signInService.getMonthlySignInCount("1001");System.out.println("当月签到次数 = " + count);// 打印当月签到详情List<String> status = signInService.getMonthlySignInStatus("1001");status.forEach(System.out::println);} finally {signInService.close();}}
}
python 版本
import redis
import datetime
import calendar
import tkinter as tk
from tkinter import messagebox, scrolledtext
# 签到逻辑（与之前一致）
class SignInService:def __init__(self, host, port):# 建立与 Redis 的连接，设置 decode_responses=True 便于直接处理字符串数据self.redis = redis.Redis(host=host, port=port, decode_responses=True)def generate_key(self, user_id, date):"""生成 Redis 的 key，格式为：user:sign:{userId}:{year}:{month}"""return f"user:sign:{user_id}:{date.year}:{date.month}"def sign_in(self, user_id):"""用户签到：- 如果今日已签到则返回 False- 否则在对应位置设置 1，并返回 True"""today = datetime.date.today()key = self.generate_key(user_id, today)day_index = today.day - 1 # Redis bitmap 下标从 0 开始if self.is_signed_in(user_id, today):return Falseself.redis.setbit(key, day_index, 1)return Truedef is_signed_in(self, user_id, date):"""检查指定日期是否已签到"""key = self.generate_key(user_id, date)day_index = date.day - 1return self.redis.getbit(key, day_index) == 1def get_monthly_sign_in_count(self, user_id):"""获取用户当月签到次数"""today = datetime.date.today()key = self.generate_key(user_id, today)return self.redis.bitcount(key)def get_monthly_sign_in_status(self, user_id):"""获取用户当月每天的签到情况，格式为：2023-10-01: 已签到/未签到"""today = datetime.date.today()key = self.generate_key(user_id, today)days_in_month = calendar.monthrange(today.year, today.month)[1]status_list = []for i in range(days_in_month):signed = self.redis.getbit(key, i)current_date = today.replace(day=i + 1)date_str = current_date.isoformat()status_list.append(f"{date_str}: {'已签到' if signed == 1 else '未签到'}")return status_listdef get_continuous_sign_in_days(self, user_id):"""获取用户连续签到天数，从今日开始向前连续统计"""today = datetime.date.today()key = self.generate_key(user_id, today)day_index = today.day - 1continuous_days = 0for i in range(day_index, -1, -1):if self.redis.getbit(key, i) == 1:continuous_days += 1else:breakreturn continuous_daysdef close(self):"""关闭 Redis 连接"""self.redis.close()
# Tkinter 界面
class SignInApp(tk.Tk):def __init__(self, sign_service, user_id):super().__init__()self.sign_service = sign_serviceself.user_id = user_idself.title("用户签到系统")self.geometry("600x500")# 签到按钮self.sign_button = tk.Button(self, text="签到", font=("微软雅黑", 14), command=self.handle_sign_in)self.sign_button.pack(pady=10)# 显示统计信息：当月签到次数、连续签到天数self.info_label = tk.Label(self, text="", font=("微软雅黑", 12))self.info_label.pack(pady=5)# 签到详情（滚动文本框）self.status_text = scrolledtext.ScrolledText(self, width=70, height=20, font=("Consolas", 10))self.status_text.pack(pady=10)# 刷新数据按钮self.refresh_button = tk.Button(self, text="刷新数据", command=self.refresh_data)self.refresh_button.pack(pady=5)# 初始化显示数据self.refresh_data()def handle_sign_in(self):# 用户点击签到时调用if self.sign_service.sign_in(self.user_id):messagebox.showinfo("签到结果", "签到成功！")else:messagebox.showwarning("签到结果", "今日已签到！")self.refresh_data()def refresh_data(self):# 更新签到统计数据count = self.sign_service.get_monthly_sign_in_count(self.user_id)continuous = self.sign_service.get_continuous_sign_in_days(self.user_id)status_list = self.sign_service.get_monthly_sign_in_status(self.user_id)self.info_label.config(text=f"当月签到次数：{count} 连续签到天数：{continuous}")self.status_text.delete("1.0", tk.END)for status in status_list:self.status_text.insert(tk.END, status + "\n")def on_close(self):# 窗口关闭时清理资源self.sign_service.close()self.destroy()
if __name__ == "__main__":# 假设用户ID为 "1001"，连接本地Redis服务器sign_service = SignInService("localhost", 6379)app = SignInApp(sign_service, "1001")app.protocol("WM_DELETE_WINDOW", app.on_close)app.mainloop()
五、总结
Bitmaps 是 Redis 处理 大规模二元状态场景 的利器，尤其适合高密度数据存储与实时统计。但在稀疏数据或需多值存储的场景中，需权衡其与集合、HyperLogLog 等其他结构的优劣。合理设计偏移量、预分配内存及集群策略，可最大化其性能优势。