什么是Redis

• Redis，远程字典服务，Remote Dirictionary Service
  1. 请求回应模式
  2. 通过tcp与redis建立连接交互
  3. 通过字典的方式索引存储数据
• 内存数据库。数据全部在内存中，不会出现需要的数据不在内存中，而在磁盘中。所以redis不能存储很多数据，通常作为辅助数据库。
• kv数据库
• 数据结构数据库，即v可以支持不同的数据结构，如string、list、hash、zset（跳表）、set、stream、hyperloglog…

安装编译

1
2
3
4
5

git clone https://gitee.com/mirrors/redis.git -b 6.2
cd redis
make 
make test
make install

启动

1
2
3
4
5
6
7
8

mkdir redis-data
# 把redis文件夹下redis.conf 拷贝到 redis-data
# 修改 redis.conf
# requirepass 修改密码 123456
# daemonize yes
cd redis-data
redis-server redis.conf # 已经启动了redis服务器
redis-cli -h 127.0.0.1 -a 123456 # 使用cli去连接redis

redis基础

string特征

• 是二进制安全的
  • 非二进制安全字符串：c语言字符串a[] = ‘hello world\0 hello’， 以\0为分割，后面的失效；
  • 二进制安全字符串：c++的string类型和protobuf中的string存储，一个string包括a[] = ‘hello world\0 hello’和len=19，包含了长度信息，后面的仍然有效；

设置及查询kv

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

# string 类型
set teacher mark 
- OK
get teacher 
- "mark"

# 散列表hash 类型，role:10001是key，name+mark是value，name是field（hash的key，需要唯一），mark是hash的value
hset role:10001 name mark 
- (integer) 1
hget role:10001 name
- "mark"

# 无序集合set类型, class是key，mark king darren 是value
sadd class mark king darren 
- (integer) 3
smembers class
- 1) "darren"
- 2) "king"
- 3) "mark"
sadd class mark
- (integer) 0 # 不会成功
sadd class vico
- (integer) 1

# 有序集合zset类型，rank是key，80 mark 90 darren 100 king是value，先写排序的值，再写set的key
# zset是跳表，就是有排序的set
zadd rank 80 mark 90 darren 100 king
- (integer) 3
# 查询0到-1（最后一个）的set值
zrange rank 0 -1
- 1) "mark"
- 2) "darren"
- 3) "king"
zadd rank 91 mark
- (integer) 0 # 改写了mark的排序依据值
zrange rank 0 -1
- 1) "darren"
- 2) "mark"
- 3) "king"

# 双端队列list类型, list是key ， mark darren king是value，是链表结构，插入有序的，lpush是从左插入
lpush list mark darren king
- (integer) 3
# 查询从头到尾，lrange表示从左到右
lrange list 0 -1
- 1) "king"
- 2) "darren"
- 3) "mark"
lpush list king 
- (integer) 4
lrange list 0 -1 # list是允许重复数据的
- 1) "king"
- 2) "king"
- 3) "darren"
- 4) "mark"

redis存储结构

• key，所有的key都是二进制安全字符串结构

redis-Value编码

• 对象类型 string、list、hash、zset、set 抽象数据结构
• 数据结构类型 对象类型底层可能是用不同的value编码 写到磁盘的结构
• 内存中的结构

string

1
2
3
4
5
6
7
8
9
10
11
12

set key 1111
- ok
object encoding key
- "int" # 说明对象类型是string，但是底层编码数据结构是int
set key 1111a
- ok
object encoding key
- "embstr" # 底层是embstr类型
set key 1555errrrrrrrrrrrrrrrrrrkkkkkkkkkkkk
- ok
object encoding key
- "raw" # 底层是raw

• 对象类型是字符串类型，是raw动态字符串，扩容规则是，
  1. 字符串长度小于1M时，加倍扩容；
  2. 超过1M每次只多扩1M；
  3. 字符串最大长度为512M；
• 二进制安全字符串，因此可以存储图片，二进制协议等二进制数据；

raw的底层存储——sds结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

struct __attribute__ ((__packed__)) sdshdr5 {
    unsigned char flags; /* 3 lsb of type, and 5 msb of string length */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr8 {
    uint8_t len; /* used */
    uint8_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr16 {
    uint16_t len; /* used */
    uint16_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr32 {
    uint32_t len; /* used */
    uint32_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr64 {
    uint64_t len; /* used */
    uint64_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};

1. 使用了柔性数组
2. 柔型数组必须在结构体，且在最后面
3. 结构体除了柔性数组至少还要有一个其他成员
4. sizeof返回结构体的大小将不包括柔性数组的大小，sizeof是编译器直接处理的，柔性数组相当于语法糖，告诉编译器这里不占用空间，但后续仍有分配空间的需求
5. malloc只需要一次
6. free也只需要一次

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

   // malloc struct sdshdr16 * p = malloc(sizeof(struct sdshdr16) + 32); // 32就是需要分配的数组大小 // free, 只需要一次 // 为什么free知道需要free的大小？？ // ans: 因为malloc是一个libc的c语言的api，返回的p的前16个字节当做头部数据，存储了sizeof(struct sdshdr16) + 32这个数据，那么free的时候会根据这个16个字节去执行free free(p); // 不使用柔性数组 struct __attribute__ ((__packed__)) sdshdr16 { uint16_t len; /* used */ uint16_t alloc; /* excluding the header and null terminator */ unsigned char flags; /* 3 lsb of type, 5 unused bits */ char *buf; }; struct sdshdr16 * p = malloc(sizeof(struct sdshdr16)); p->buf = malloc(32); // 需要额外malloc free(p->buf); // 也需要额外free free(p);

• 柔性数组的优点：
  1. 方便内存分配和释放
  2. 减少内存碎片

基础命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 设置 key 的 value 值
SET key val
# 获取 key 的 value
GET key
# 执行原子加一的操作
INCR key # 为何需要原子操作？因为它可以将查询和修改两个网络交互合并成一个并返回结果，减少了网络交互
# 执行原子加一个整数的操作
INCRBY key increment
# 执行原子减一的操作
DECR key
# 执行原子减一个整数的操作
DECRBY key decrement
# 如果key不存在，这种情况下等同SET命令。 当key存在时，什么也不做
# set Not eXist ok 这个命令是否执行了 0,1 是不是操作结果是不是成功
SETNX key value
# 删除 key val 键值对
DEL key
# 设置或者清空key的value(字符串)在offset处的bit值。
setbit embstr raw int
# 动态字符串 能够节约内存
SETBIT key offset value
# 返回key对应的string在offset处的bit值
GETBIT key offset
# 统计字符串被设置为1的bit数.
BITCOUNT key

存储结构

• 字符串长度小于等于 20 且能转成整数，则使用 int 存储；
• 字符串长度小于等于 44，则使用 embstr 存储；
• 字符串长度大于 44，则使用 raw 存储；

应用

1. 对象存储

   1 2 3 4 5 6 7 8

   SET role:10001 '{["name"]:"mark",["sex"]:"male",["age"]:30}' SET role:10002 '{["name"]:"darren",["sex"]:"male",["age"]:30}' # 极少修改，对象属性字段很少改变的时候 GET role:10001 # key 如何来设置 # 1. 有意义的字段 role 有多行 # 2. role:10001 redis 客户端 role:10001:recharge role:10001:activity:10001

2. 累加器

   1 2 3 4

   # 统计阅读数 累计加1 incr reads # 累计加100 incrby reads 100

3. 分布式锁

   1 2 3 4 5 6 7 8

   # redis 实现是 非公平锁 setnx lock 1 # 不存在才能设置 定义加锁行为 占用锁 setnx lock uuid # expire 30 过期 set lock uuid nx ex 30 # 释放锁 del lock if (get(lock) == uuid) del(lock);

4. 位运算

   1 2 3 4 5 6 7

   # 猜测一下 string 是用的 int 类型 还是 string 类型 # 月签到功能 10001 用户id 202106 2021年6月份的签到 6月份的第1天 setbit sign:10001:202106 1 1 # bit的意思是存储的值是bit类型，只有1或0 # 计算 2021年6月份 的签到情况 bitcount sign:10001:202106 # 获取 2021年6月份 第二天的签到情况 1 已签到 0 没有签到 getbit sign:10001:202106 2

list

基础命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 从队列的左侧入队一个或多个元素
LPUSH key value [value ...]
# 从队列的左侧弹出一个元素
LPOP key
# 从队列的右侧入队一个或多个元素
RPUSH key value [value ...]
# 从队列的右侧弹出一个元素
RPOP key
# 返回从队列的 start 和 end 之间的元素 0, 1 2 负索引
LRANGE key start end
# 从存于 key 的列表里移除前 count 次出现的值为 value 的元素
# list 没有去重功能 hash set zset
LREM key count value
# 它是 RPOP 的阻塞版本，因为这个命令会在给定list无法弹出任何元素的时候阻塞连接
BRPOP key timeout # 超时时间 + 延时队列

应用

1. 栈

   1 2 3

   LPUSH + LPOP # 或者 RPUSH + RPOP

2. 队列

   1 2 3

   LPUSH + RPOP # 或者 RPUSH + LPOP

3. 阻塞队列

   1 2 3

   LPUSH + BRPOP # 或者 RPUSH + BLPOP

4. 异步消息队列

   1	# 操作与队列一样，但是在不同系统间；生成者和消费者；

5. 获取固定窗口记录

   1 2 3 4 5 6 7 8 9 10

   # 在某些业务场景下，需要获取固定数量的记录；比如获取最近50条战绩；这些记录需要按照插入的先后顺序返回； lpush says '{["name"]:"mark",["text"]:"祝大家儿童节快乐!", ["picture"]:["url://image-20210601172741434.jpg","url://image-20210601172741435.jpg"], timestamp =1231231230}' lpush says '{["name"]:"king",["text"]:"祝大家儿童节快乐!", ["picture"]:["url://image-20210601172742434.jpg","url://image-20210601172741436.jpg"], timestamp =1231231231}' lpush says '{["name"]:"darren",["text"]:"祝大家儿童节快乐!", ["picture"]:["url://image-20210601172743434.jpg","url://image-20210601172741437.jpg"], timestamp =1231231232}' lpush says '{["name"]:"mark",["text"]:"一切只为渴望更优秀的你", ["picture"]:["url://image-20210601172744434.jpg","url://image-20210601172741438.jpg"], timestamp =1231231233}' lpush says '{["name"]:"darren",["text"]:"hello 0Voice! hello to better self",["picture"]:["url://image-20210601172745439.jpg","url://image-20210601172741435.jpg"], timestamp =1231231234}' lpush says '{["name"]:"king",["text"]:"真牛逼！", ["picture"]:["url://image-20210601172745434.jpg","url://image-20210601172741440.jpg"], timestamp =1231231235}' # 裁剪最近5条记录 战绩 近50条 ltrim says 0 4 lrange says 0 -1

• 实际项目中需要保证命令的原子性，所以一般用 lua 脚本 或者 使用pipeline 命令；

  1 2 3 4

  -- redis lua脚本 local record = KEYS[1] redis.call("LPUSH", "says", record) redis.call("LTRIM", "says", 0, 4)

hash

基础命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 获取 key 对应 hash 中的 field 对应的值
HGET key field
# 设置 key 对应 hash 中的 field 对应的值
HSET key field value
# 设置多个hash键值对
HMSET key field1 value1 field2 value2 ... fieldn
valuen
# 获取多个field的值
HMGET key field1 field2 ... fieldn
# 给 key 对应 hash 中的 field 对应的值加一个整数值
HINCRBY key field increment
# 获取 key 对应的 hash 有多少个键值对
HLEN key
# 删除 key 对应的 hash 的键值对，该键为field
HDEL key field

存储结构

• 节点数量大于 512（hash-max-ziplist-entries） 或所有字符串长度大于 64（hash-max-ziplist-value），则使用 dict 实现；
• 节点数量小于等于 512 且有一个字符串长度小于 64，则使用ziplist 实现；

应用

1. 存储对象

   1 2 3 4 5 6 7 8 9 10 11

   hmset hash:10001 name mark age 18 sex male # 与 string 比较 set hash:10001 '{["name"]:"mark",["sex"]:"male",["age"]:18}' # 假设现在修改 mark的年龄为19岁 # hash： hset hash:10001 age 19 # string: get role:10001 # 将得到的字符串调用json解密，取出字段，修改 age 值 # 再调用json加密 set role:10001 '{["name"]:"mark",["sex"]:"male",["age"]:19}'

2. 购物车

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

   # 将用户id作为 key # 商品id作为 field # 商品数量作为 value # 注意：这些物品是按照我们添加顺序来显示的； # 添加商品： hset MyCart:10001 40001 1 lpush MyItem:10001 40001 # 增加数量： hincrby MyCart:10001 40001 1 hincrby MyCart:10001 40001 -1 // 减少数量1 # 显示所有物品数量： hlen MyCart:10001 # 删除商品： hdel MyCart:10001 40001 lrem MyItem:10001 1 40001 # 获取所有物品： lrange MyItem:10001 # 40001 40002 40003 hget MyCart:10001 40001 hget MyCart:10001 40002 hget MyCart:10001 40003

set

基础命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 添加一个或多个指定的member元素到集合的 key中
SADD key member [member ...]
# 计算集合元素个数
SCARD key
# SMEMBERS key
SMEMBERS key
# 返回成员 member 是否是存储的集合 key的成员
SISMEMBER key member
# 随机返回key集合中的一个或者多个元素，不删除这些元素
SRANDMEMBER key [count]
# 从存储在key的集合中移除并返回一个或多个随机元素
SPOP key [count]
# 返回一个集合与给定集合的差集的元素
SDIFF key [key ...]
# 返回指定所有的集合的成员的交集
SINTER key [key ...]
# 返回给定的多个集合的并集中的所有成员
SUNION key [key ...]

存储结构

• 元素都为整数且节点数量小于等于 512（set-max-intsetentries），则使用整数数组存储;
• 元素当中有一个不是整数或者节点数量大于 512，则使用字典存储；

应用

1. 抽奖

   1 2 3 4 5 6 7 8

   # 添加抽奖用户 sadd Award:1 10001 10002 10003 10004 10005 10006 sadd Award:1 10009 # 查看所有抽奖用户 smembers Award:1 # 抽取多名幸运用户 srandmember Award:1 10 # 如果抽取一等奖1名，二等奖2名，三等奖3名，该如何操作？

2. 共同关注

   1 2 3

   sadd follow:A mark king darren mole vico sadd follow:C mark king darren sinter follow:A follow:C

3. 推荐好友

   1 2 3 4

   sadd follow:A mark king darren mole vico sadd follow:C mark king darren # C可能认识的人： sdiff follow:A follow:C

zset

基础命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 添加到键为key有序集合（sorted set）里面
ZADD key [NX|XX] [CH] [INCR] score member [score member ...]
# 从键为key有序集合中删除 member 的键值对
ZREM key member [member ...]
# 返回有序集key中，成员member的score值
ZSCORE key member
## 为有序集key的成员member的score值加上增量increment
ZINCRBY key increment member
# 返回key的有序集元素个数
ZCARD key
# 返回有序集key中成员member的排名
ZRANK key member
# 返回存储在有序集合key中的指定范围的元素 order by id
limit 1,100
ZRANGE key start stop [WITHSCORES]
# 返回有序集key中，指定区间内的成员(逆序)
ZREVRANGE key start stop [WITHSCORES]

存储结构

• 节点数量大于 128 或者有一个字符串长度大于 64，则使用跳表（skiplist）；
• 节点数量小于等于 128（zset-max-ziplist-entries）且所有字符串长度小于等于 64（zset-max-ziplist-value），则使用ziplist 存储；
• 数据少的时候，节省空间，O(n)
• 数据大的时候，访问性能，O(1)或O(logn)

应用

1. 搜索热榜

   1 2 3 4 5 6 7 8 9 10 11 12 13

   # 点击新闻： zincrby hot:20230612 1 10001 zincrby hot:20230612 1 10002 zincrby hot:20230612 1 10003 zincrby hot:20230612 1 10004 zincrby hot:20230612 1 10005 zincrby hot:20230612 1 10006 zincrby hot:20230612 1 10007 zincrby hot:20230612 1 10008 zincrby hot:20230612 1 10009 zincrby hot:20230612 1 10010 # 获取排行榜： zrevrange hot:20230612 0 9 withscores

2. 延时队列

• 将消息序列化成一个字符串作为 zset 的 member；这个消息的到期处理时间作为 score，然后用多个线程轮询 zset 获取到期的任务进行处理

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

  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) # 使用连接池 def loop(): while True: values = redis.zrangebyscore("delayqueue", 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) # 缺点：loop 是多线程竞争，两个线程都从zrangebyscore获取到数据，但是zrem一个成功一个失败， # 优化：为了避免多余的操作，可以使用lua脚本原子执行这两个命令 # 解决：漏斗限流

1. 分布式定时器

   1	# 生产者将定时任务 hash 到不同的 redis 实体中，为每一个redis 实体分配一个 dispatcher 进程，用来定时获取 redis 中超时事件并发布到不同的消费者中；

2. 时间窗口限流

• 系统限定用户的某个行为在指定的时间范围内（动态）只能发生 N 次；

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

  ## 指定用户 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() -- 记录行为 red:zadd(key, now, now) -- 移除时间窗口之前的行为记录，剩下的都是时间窗口内的记录 red:zremrangebyscore(key, 0, now - period*100) -- 获取时间窗口内的行为数量 red:zcard(key) -- 设置过期时间，避免冷用户持续占用内存 时间窗口的长度+1秒 red:expire(key, period + 1) local res = red:commit_pipeline() return res[3] <= max_count end # 维护一次时间窗口，将窗口外的记录全部清理掉，只保留窗口内的记录； # 缺点：记录了所有时间窗口内的数据，如果这个量很大，不适合做这样的限流；漏斗限流 # 注意：如果用 key + expire 操作也能实现，但是实现的是熔断限流，这里是时间窗口限流的功能；

怎么设计kv

怎么设计key

• 单个功能一个key，取有意义的名字
• 相同功能多个key，以：作为分割（以为客户端工具可以根据：生成树状结构）

怎么设计value

• 数据特性

redis抽象层次

• 关注具体返回值的意义
• redis没有创建数据结构的命令
  1. 设置的同时创建
  2. 添加的同时创建
• redis有删除kv的命令以及当v中没有元素的时候会自动删除
• 阻塞连接的概念
• 通过命令的组合实现其他数据结构
• 通过组合数据结构实现功能
  1. hash+list
  2. hash+set
  3. hash+zset