Redis（Remote Dictionary Server，远程词典服务）是一种基于内存的数据库，数据的读写操作都是在内存中完成，读写速度快，常用于缓存、消息队列、分布式锁场景。
1、Redis 基础
1.1、Redis 的特点
- 远程字典服务
- 节点：通过 tcp 与 redis 建立连接交互，redis 远程字典类似
unordered_map<string, T> - 请求回应模型：命令请求 + 返回结果
- 内存数据库：数据读写操作都在内存中。内存数据持久化到磁盘中，支持数据持久化
- KV 数据库：KV 存储，key 都是 string 类型，value 提供了丰富的数据结构
- 单线程，每个命令具备原子性，不存在并发竞争
- 支持主从集群、分片集群
- 低延迟，速度快（基于内存、IO多路复用、良好的编码）
1.2、* NoSQL
SQL 关系型数据库，NoSQL 非关系型数据库。Redis 属于 NoSQL。
NoSQL 数据库的分类
- KV 数据库：Redis
- 列存储数据库：Cassandra, HBase
- 文档型数据库：MongoDb
- 图数据库：Neo4J, Infinite Graph
SQL vs NoSQL
- 数据结构：结构化 vs 非结构化。
- 数据关联：关联的 vs 无关联的。
- 查询方式：SQL vs 非 SQL
- 事务特性：ACID vs BASE
事务特性
SQL 遵循 ACID 原则
- 原子性：一个事务中的操作，要么都做，要么都不做，不可分割。
- 一致性：事务的前后，数据满足完整性约束，数据库保持一致性状态。
- 独立性：并发事务间相互隔离，互不影响。
- 持久性：事务一旦提交，其结果就是永久性的。
NoSQL 遵循 BASE 原则
- 基本可用：数据在大多数状态可用，并分布在不同的机器上
- 软状态：副本并不总是一致的
- 最终一致性：数据在某一时间点保持一致，但不能保证何时一致
使用场景
- SQL：安全优先，保证数据一致性
- NoSQL：效率优先，高性能、高可用性、可伸缩性，没有复杂的关系
1.3、* Redis vs Memcached
Redis 和 Memcached 都是内存型数据库，数据保存在内存中，通过 tcp 直接存取，高性能，高并发，一般用来作为缓存。
Redis 优点
- Redis 支持的数据结构更丰富；Memcached 仅支持 KV 数据类型
- Redis 支持的数据持久化；Memcached 不支持持久化，数据全部在内存中，宕机后，数据全部丢失。
- Redis 原生支持集群模式；Memcached 没有原生的集群模式
- Redis 支持订阅模型、事务、lua 脚本等多种功能；Memcached 就是一个简单的 KV 缓存
Memcached 优点
- 多核优势，单实例吞吐量极高，适用于最大程度抗量，为服务器减压。大多数公司使用它。
2 、Redis 配置
安装编译
# 安装 redis-6.2.7 wget https://download.redis.io/releases/redis-6.2.7.tar.gz tar zxvf redis-6.2.7.tar.gz cd redis-6.2.7/ make make test make install # 默认安装路径 /usr/local/bin # 安装 hiredis cd /deps/hiredis make make test make install # 问题：You need tcl 8.5 or newer in order to run the Redis test wget http://downloads.sourceforge.net/tcl/tcl8.6.1-src.tar.gz tar xzvf tcl8.6.1-src.tar.gz cd tcl8.6.1/unix/ ./configure make make install
前台启动
# 前台启动，卡在当前界面，退出则关闭 redis redis-server
后台启动
修改 redis.conf 配置文件
# 在 redis 文件夹下，备份 redis.conf mkdir redis-data cp redis.conf redis-data/ cd redis-data # 修改 redis.conf # 监听的地址，默认是127.0.0.1，会导致只能在本地访问。修改为0.0.0.0则可以在任意IP访问，生产环境不要设置为0.0.0.0 bind 0.0.0.0 # 守护进程，修改为yes后即可后台运行 daemonize yes # 密码，设置后访问 redis 必须输入密码: 命令auth 123456 # requirepass 123456 # 其他修改（可选） # 监听的端口 port 6379 # 工作目录，默认当前目录，运行redis-server时的命令，日志、持久化等文件保存在这里 dir . # 数据库数量，设置为1，代表只使用1个库，默认有16个库，编号0~15 databases 1 # 设置 redis 能够使用的最大内存 maxmemory 512mb # 日志文件，默认为空，不记录日志，可以指定日志文件名 logfile "redis.log"
指定配置文件后台启动
# 指定配置文件后台启动 redis-server redis.conf # 查看 redis 进程 ps aux | grep redis
开机启动
# 创建系统服务文件 vim /etc/systemd/system/redis.service # 写入命令 [Service] # 开机启动 ExecStart=/opt/redis-6.2.7/redis-data/src/redis-server /opt/redis-6.2.7/redis-data/redis.conf # 重载系统服务 systemctl daemon-reload # 配置生效 systemctl enable redis # 启动 systemctl start redis # 查看状态 systemctl status redis # 重启 systemctl restart redis # 停止 systemctl stop redis
客户端连接
redis-cli [options] [commonds] -h 指定要连接的redis节点的IP地址，默认是127.0.0.1 -p 指定要连接的redis节点的端口，默认是6379 -a 指定redis的访问密码
3、Redis 命令
Redis 命令官方文档：redis Commands。数据结构的原理，见我之前的博客：Redis 数据结构
3.1、String
字符串值的索引
- 正数索引从 0 开始， 从字符串开头向结尾不断递增
0 1 2 3 ... n - 负数索引从 -1 开始，从字符串结尾向开头不断递减
-n ...-3 -2 -1
常用命令
# 设置字符串键的值 SET key val # 获取字符串键的值 GET key # 获取旧值并设置新值 GET SET # 一次为多个字符串键设置值 MSET key value # 一次获取多个字符串键的值 MGET key [key...] # set Not eXist。若 key 存在等同于 SET；若 key 存在，什么也不做 SETNX key value MSETNC key value [key value ...] # 获取字符串值指定索引范围上的内容 GETRANGE key start end # 对字符串值的指定索引范围进行设置 SETRANGE key start end # 追加新内容到值的末尾 APPEND key suffix # 执行原子+1的操作 INCR key # 执行原子加一个整数的操作 INCRBY key increment # 执行原子加一个浮点数的操作 INCRBYFLOAT key increment # 执行原子-1的操作 DECR key # 执行原子减一个整数的操作 DECR key decrement # 删除字符串键值对 DEL key # 二进制安全字符串，可以基于此做位运算 # 设置字符串键在offset处的bit值 SETBIT key offset value # 获取字符串键在offset处的bit值 GETBIT key offset # 统计字符串设置为1的bit数. BITCOUNT key
应用实例
例1：对象存储，适用于对象属性字段极少修改
set role:1001 'name:mark,sex:male,age:30' get role:1001
例2：累加器
# 统计阅读数 incr reads incrby reads 100
例3：分布式锁，redis 实现是非公平锁
# 加锁 set lock uuid nx ex 30 # 解锁 if (get(lock) == uuid) del(lock);
例4：位运算实现月签到功能
# 实现月签到功能 # 签到 用户id 年月 日 是否签到 setbit sign:1001:202210 1 1 # 2022年10月1日 签到 setbit sign:1001:202210 2 0 # 2022年10月2日 没签 # 获取该用户2022年10月份签到次数 bitcount sign:1001:202210 # 获取该用户2022年10月1日是否签到 getbit sign:1001:202210 1
3.2、List
双向链表实现，首尾操作（增删）时间复杂度O(1)，查找元素时间复杂度O(n)。
常用命令
# 将元素推入列表 LPUSH list value [value ...] RPUSH list value [value ...] # 弹出列表元素 LPOP list RPOP list # 获取列表长度 LLEN list # 获取指定索引上的元素 LINDEX list index # 获取指定索引范围上的元素 LRANGE list start end # 为指定索引设置新元素 LSET list index new_element # 将元素插入列表 LINSERT list BEFORE|AFTER target_element new_element # 修剪列表，移除范围之外的所有元素，保留给定范围的元素 LTREM list count value # 移除列表中的指定元素，count=0，移除列表中所有指定元素；count>0，移除从列表左端开始前count个指定元素，count<0，移除从列表右端开始前count个指定元素 LREM list count element # 阻塞式弹出 BLPOP list [list...] timeout # 延时队列 + 超时时间 BRPOP list [list...] timeout # 延时队列 + 超时时间
应用实例
例1：数据结构
# 栈 LPUSH + LPOP RPUSH + RPOP # 队列 LPUSH + RPOP RPUSH + LPOP # 阻塞队列，异步消息队列 LPUSH + BRPOP RPUSH + BLPOP
例2：获取固定窗口记录
# 查询战绩 lpush win 'k:0,d:11,a:0' lpush win 'k:0,d:13,a:5 ... # 裁剪最近5条记录，实际项目保证命令的原子性，一般用 lua 脚本或 pipline 命令 ltrim win 0 4 lrange win 0 -1
3.3、Hash
散列，查询修改 O(1)
常用命令
# 设置散列中字段的值 HSET hash field value # 只在字段不存在的情况下为它设置值 HSETNX hash field value # 获取字段的值 HGET hash field # 一次为多个字段设置值 HMSET hash key field value [field value...] # 一次获取多个字段的值 HMGET hash field [field...] # 对字段存储的整数值执行加法或减法操作 HINCRBY hash field increment # 对字段存储的数字值执行浮点数加法或减法操作 HINCRBYFLOAT hash field increment # 获取散列包含的字段数量 HLEN hash # 获取字段值的字节长度 HSTRLEN hash # 检查字段是否存在 HEXISTS hash field # 删除字段 HDEL hash field
应用场景
hash 的用途广泛，可以存储需要频繁修改的对象。若为 string 类型首先把获得的字符串 json 反序列化，修改后，再用 json 序列化，操作繁琐。
也可以通过组合数据结构实现不同的功能
hash(存储对象) + list（存储插入的顺序） hash（存储对象）+ set（所有在线的玩家） hash + zset (排行榜)
例：购物车：hash + list
# hash: 管理购物车中商品的数量，用户id作为hash，商品id作为field，商品数量作为value # list: 存储购物车中的商品，按照添加顺序来显示的 # 添加商品 hset MyCart:1001 4001 1 lpush MyItem:1001 4001 # 增加或减少购物车中的商品数量 hincrby MyCart:1001 4001 1 hincrby MyCart:1001 4001 -1 # 删除购物车中的商品 hdel MyCart:1001 4001 lrem MyItem:1001 1 4001 # 显示购物车中所有商品数量 hlen MyCart:1001 # 获取购物车中所有商品 lrange MyItem:1001 0 -1 # 获取每个商品的数量 hget Mycart:1001 4002 hget MyCart:1001 4003
3.4、set
无序集合，无序唯一。
常用命令
# 将元素添加到集合 SADD set lement [element ...] # 从集合中移除元素 SREM set element [element ...] # 将元素从一个集合移动到另一个集合 SMOVE source target element # 获取集合包含的所有元素 SMEMBERS set # 获取集合包含的元素数量 SCARD set # 检查给定元素是否存在于集合 SISMEMBER set element # 随机获取集合中的元素 SRANDMEMBER set [count] # 随机从集合中移除指定数量的元素 SPOP set [count] # 集合交集运算 SINTER set [set...] SINTERSTORE key set [set...] # 交集运算结构存储到指定的键里 # 集合并集运算 SUNION set [set...] SUNIONSTORE key set [set...] # 并集运算结构存储到指定的键里 # 集合差集运算 SDIFF set [set...] SDIFFSTORE key set [set...] # 并集运算结构存储到指定的键里
应用场景
例1：共同关注，推荐好友
sadd follow:A liubei guanyu zhangfei sadd follow:B guanyu zhaoyun weiyan # 获取共同关注 sinter follow:A follow:B # 向 B 推荐 A 的好友 sdiff follow:A follow:B # 向 A 推荐 B 的好友 sdiff follow:B follow:A
例2：抽奖
# 添加抽奖用户 sadd Award 0 1 2 3 4 5 6 7 8 9 # 查看抽奖用户 127.0.0.1:6379> smembers Award # 抽取多名获奖用户 srandmember Award 1 # 抽取一等奖1名，二等奖2名，三等奖3名 spop Award 1 spop Award 2 spop Award 3
3.5、zset
有序集合，有序唯一。
常用命令
# 添加或更新成员 ZADD score_set socre member [score member ...] # 移除指定的成员 ZREM score_set member [member ...] # 获取成员的分值 ZSCORE score_set member # 对成员的分值执行自增或自减操作 ZINCRBY score_set increment member # 获取有序集合的大小 ZCARD key # 获取成员在有序集合中的排名（升序或降序） ZRANK score_set member ZREVRANK score_set member # 获取指定索引范围内的成员（升序或降序） ZRANGE score_set start stop [WITHSCORES] ZREVRANGE score_set start stop [WITHSCORES] # 获取指定分值范围内的成员（升序或降序） ZRANGEBYSCORE score_set start stop [WITHSCORES] ZREVRANGEBYSCORE score_set start stop [WITHSCORES] # 统计指定分值范围内的成员数量 ZCOUNT score_set min max # 移除指定排名范围内的成员 ZREMRANGEBYRANK score_set start end # 移除指定分值范围内的成员 ZREMRANGEBYSCORE score_set start end
应用场景
例1：热搜排行榜
# 添加热搜id + 点击量 zadd hot:20221015 NX 0 1001 0 1002 0 1003 # 点击热搜 zincrby hot:20221015 1 1001 zincrby hot:20221015 1 1001 zincrby hot:20221015 1 1002 # 获取热搜排行榜 zrevrange hot:20221015 0 -1 withscores
例2：延时队列 *
将消息序列化成一个字符串作为 zset 的 member，消息的到期时间作为 score，然后用多个线程轮询 zset 获取到期的任务进行处理。
-- 添加延时消息 def delay(msg): msg.id = str(uuid.uuid4()) -- 保证 member 唯一 value = json.dumps(msg) retry_ts = time.time() + 5 -- 当前时间 + 5s后重试 redis.zadd("delay-queue", retry_ts, value) -- 多线程获取消息，都可以成功 zrangebyscore 获取数据，但只有一个能 zrem 成功 -- 优化：可以使用 lua 脚本原子执行这两个命令 def loop(): while True: -- 获取到期的消息 values = redis.zrangebyscore("delay-queue", 0, time.time(), start=0, num=1) if not values: time.sleep(1) continue value = values[0] -- 从延时队列中移除过期消息，并执行 success = redis.zrem("delay-queue", value) if success: msg = json.loads(value) handle_msg(msg)
例3：分布式定时器
生产者将定时任务 hash 到不同的 redis 实体中，为每一个redis 实体分配一个 dispatcher 进程，用来定时获取 redis 中超时事件并发布到不同的消费者中
例4：时间窗口限流
限制单个接口在一定时间内的请求次数，
-- 指定用户 user_id 的某个行为 action 在特定时间范围内 period 只允许发生 max_count 次 -- 维护一次时间窗口，将窗口外的记录全部清除掉，只保留窗口内的记录 local function is_action_allowed(red, userid, action, period, max_count) local key = tab_concat({"hist", userid, action}, ":") local now = zv.time() -- 开启管道 red:init_pipeline() -- 记录行为：关键是 score，member 无意义 red:zadd(key, now, now) -- 移除时间窗口之前的行为记录，剩下的都是时间窗口内的记录 red:zremrangebyscore(key, 0, now - period) -- 获取时间窗口内的行为数量 red:zcard(key) -- 设置过期时间，避免冷用户持续占用内存 时间窗口长度 + 1秒 red:expire(key, period + 1) -- 提交管道内的命令 local res = red:commit_pipeline() -- 不超过次数返回 ture，超过返回 false return res[3] <= max_count end
