Redis基于项目方案的结构功能解析

发表于 2022-05-22 分类于开发资料阅读次数：

Redis结构功能解析

Remote Dictionary Server , 一款开源, 使用key-value存储的内存型数据结构服务器

简介

NoSQL

	SQL数据库	NoSQL数据库
数据结构	结构化(Structured)	非结构化
数据关联	关联(Relational)	无关联
查询方式	SQL查询	非SQL
事务特性	ACID	BASE
存储方式	磁盘	内存
扩展性	垂直	水平
使用场景	1) 数据结构固定 2) 业务对数据安全性、一致性要求较高	1) 数据结构不固定 2) 对一致性、安全性要求不高 3) 对性能需求优先级高

特征

键值(key-value)型, value支持多种不同数据结构, 功能丰富
工作线程为单线程, 每个命令具备原子性
低延迟、速度快(基于内存、I/O多路复用、良好编码)
支持数据持久化
支持主从集群、分片集群
支持多语言客户端

安装

apt安装

$ sudo apt install redis	# 安装
$ ps aux | grep redis		# 查看进程
$ sudo service redis-server stop	# 停止服务
$ sudo apt remove redis-server		# 移除安装

编译安装

# 安装编译器
$ sudo apt install make gcc tcl
# 下载源码包
$ wget https://github.com/redis/redis/archive/refs/tags/6.2.6.tar.gz
# 解压并移动到指定目录
$ tar -zxvf 6.2.6.tar.gz
$ sudo mv redis-6.2.6 /usr/local/src/
$ cd /usr/local/src/redis-6.2.6
# 编译/测试
$ make && make install
$ make test

配置

# 备份原有配置文件
$ cp /usr/local/src/redis-6.2.6/redis.conf /usr/local/src/redis-6.2.6/redis.conf.bak
# 修改配置文件
$ vim /usr/local/src/redis-6.2.6/redis.conf

# bind 127.0.0.1 -::1	# 修改为允许任意IP访问
bind 0.0.0.0 -::1

# daemonize no			# 修改为守护进程运行
daemonize yes

# requirepass foobared	# 修改为需要密码
requirepass xxxxxxxxx

# port 6379				# 修改默认端口
port 16379

# dir ./				# 修改工作目录
dir /usr/local/src/redis-6.2.6/work/

# databases 16			# 修改数据库数量

# maxmemory <bytes>		# 设置能够使用的最大内存

# logfile ""			# 指定日志文件名,不设置则标准输出,守护进程则废弃日志
logfile "/var/log/redis.log"

# 保护模式，关闭保护模式
protected-mode no

# 数据库数量，设置为1
databases 1

运行

1
2
3

$ cd /usr/local/src/redis-6.2.6
# 指定配置文件运行
$ redis-server redis.conf

关闭

1	$ redis-cli -a xxxxxxxxx -p 16379 shutdown

配置systemctl

1	$ sudo vi /etc/systemd/system/redis.service

[Unit]
Description=redis-server
After=network.target

[Service]
Type=forking
ExecStart=/usr/local/bin/redis-server /usr/local/src/redis-6.2.6/redis.conf
PrivateTmp=true

[Install]
WantedBy=multi-user.target

$ sudo systemctl daemon-reload 	# 载入
$ sudo systemctl start redis			# 启动Redis服务
$ sudo systemctl stop redis			# 停止Redis服务
$ sudo systemctl restart redis	 	# 重启Redis服务
# 若报错: 
# Failed to start redis.service: Unit redis-server.service is masked.
$ sudo systemctl unmask redis-server.service

通过systemctl配置开机自启动

$ sudo systemctl enable redis
# 若报错: 
# Failed to enable unit: Refusing to operate on linked unit file redis.service

$ sudo systemctl is-enabled redis	# 检查是否已存在自启动文件
$ sudo systemctl disable redis		# 关闭原有自启动文件
$ sudo vi /etc/systemd/system/redis.service	# 重新写入文件
$ sudo systemctl enable redis		# 开启自启动

客户端

命令行客户端

1	$ redis-cli [options] [command]

options:
	-h 192.168.253.138: 指定需要连接的redis节点的IP地址, 默认为127.0.0.1
	-p 16379: 指定需要连接的redis节点的端口, 默认为6379
	-a xxxxxx: 指定redis的访问密码
command:
	需要redis服务器执行的操作命令, 若不指定, 进入交互控制台

可以在连接服务时不指定密码, 进入控制台后使用auth命令补充密码

图形化桌面客户端

AnotherRedisDesktopManager

1	https://github.com/qishibo/AnotherRedisDesktopManager

数据结构

key-value数据库, key一般为String, value支持多种数据类型:

数据类型	示例
字符串 - String	‘hello redis’
哈希 - Hash	{name: ‘jack’, age: 21}
列表 - List	[A -> B -> C -> C]
集合 - Set	{A, B, C}
有序集合 - SortedSet	{A: 1, B: 2, C: 3}
GEO	{A: (120.3, 30.3)}
BitMap	0110110101110101011
HyperLogLog	0110110101110101011

1
2
3

数据结构对应命令可通过官方文档或客户端help @[数据类型]查看
 	- https://redis.io/commands/
	- help @string

通用命令

1	文档: [Redis Generic Commands](https://redis.io/commands/?group=generic)

keys	查看符合模板pattern的所有key, 可能耗费较多系统资源
	支持的pattern模式
	- h?llo		匹配hello，hallo和hxllo
    - h*llo		匹配hllo和heeeello
    - h[ae]llo	匹配hello但hallo,不匹配hillo
    - h[^e]llo	匹配hallo, hbllo, ... 但不匹配hello
    - h[a-b]llo	匹配hallo和hbllo
del 	删除指定的键。如果键不存在，则忽略它. 返回值为已删除成功的键数量
copy	复制指定键内容到指定键
rename	重命名键名
exists 	判断指定key是否存在
expire 	为一个key设置有效期, 过期自动删除
ttl 	查看一个key的过期剩余时间, -1代表永不过期, -2代表key已不存在

String类型命令

字符串类型包括: string普通字符串, int整数类型, 可以自增减, float浮点类型, 可以自增减

最大空间不超过512m

set			添加或修改一个字符串类型的键值对
get			获取指定key对应的值
getdel 	 	获取key对应的值后删除key
mset  		批量设置键值对
mget 		获取多个key对应的值
incr 		为一个整型key对应的值自增1
incrby 	 	为一个整型key对应的值以指定步长自增
incrbyfloat 为一个浮点型key对应的值以指定步长自增
decr 		为一个整型key对应的值自减1
decrby 		为一个整型key对应的值以指定步长自减
setnx		仅在指定key不存在时才执行set操作, 相当于set key value nx
msetnx		仅在指定key不存在时才执行多个set操作
setex 		设置指定key-value并指定过期时间, 相当于set key value ex 10
psetex		使用微秒作为单位设置指定key-value并指定过期时间
setrange 	指定从偏移位置起, 覆写原有字符串
getrange 	指定起始-结束获取子字符串, 相当于 substr key 0 2
strlen 		获取key对应值的字符串长度

存储多项目, 多类型, 多用户时, 可以使用符号 : 格式设置键名

mset 
project1:module1:user1 v1 
project1:module2:user1 v2 
project1:module1:user2 v3  
project1:module2:user2 v1

Hash类型命令

哈希类型, 也叫散列, 其value为一个无序字典, 可以将对象中每个字段独立存储, 单独做CURD

hset		添加或修改hash类型key的值
hget		获取一个hash类型key的field对应的值
hmset		批量添加多个hash类型key的field对应的值
hmget		批量获取多个hash类型key的field对应的值
hgetall		获取一个hash类型key的所有field对应的值
hkeys		获取一个hash类型key中的所有fields
hvals		获取一个hash类型key中的所有fields对应的值
hincrby		使一个hash类型key的field对应的整型值自增并指定步长
hsetnx		仅在不存在时, 添加一个hash类型key的field对应的值
hlen		获取一个hash类型key中field的数量
hstrlen		获取一个hash类型key中field对应值的长度

List类型命令

可以看做双向链表, 即支持正向检索, 也支持反向检索

特征: 有序, 元素可重复, 插入/删除比较快, 查询速度一般

一般用于存储有序数据, 包括点赞列表, 评论列表等

lpush		向列表左侧插入一个或多个元素
lpop		移除并返回左侧第一个元素, 没有则返回nil
rpush		向列表右侧插入一个或多个元素
rpop		移除并返回右侧第一个元素, 没有则返回nil
lrange		返回一段索引范围内的所有元素
blpop		阻塞式移除并返回左侧第一个元素, 无元素则等待至超时
brpop		阻塞式移除并返回右侧第一个元素, 无元素则等待至超时

模拟栈(先进后出, 即入口出口在同一边)

rpush + rpop / lpush + lpop

模拟队列(先进先出, 即入口出口在不同边)

rpush + lpop / lpush + rpop

模拟阻塞队列(先进先出, 无则不出, 即入口出口在不同边, 出队时采用blpop或brpop)

rpush + blpop / lpush + brpop

Set类型命令

特征: 无序, 元素不可重复, 查找快, 支持交并差集操作

一般用于好友列表/ 共同关注/ 好友关系等

1
2
3

数据结构对应命令可通过官方文档或客户端help @[数据类型]查看
 	- https://redis.io/commands/
	- help @set

sadd key member 	向集合中添加一个或多个元素
srem key member		移除集合中的指定元素
scard key			返回集合中元素的个数
sismember key member	判断某个元素是否存在于集合中
smembers key		获取集合中的所有元素
sinter key1 key2	获取集合key1和key2的交集
sdiff key1 key2		获取集合key1与key2的差集
dunion key1 key2	获取集合key1与key2的并集

SortedSet类型命令

可排序集合, 集合中的每一个元素都带有一个score属性, 基于score实现元素排序

特性: 可排序 , 元素不可重复, 查询速度快

基于可排序性 , 一般用于排行榜功能

1
2
3

数据结构对应命令可通过官方文档或客户端help @[数据类型]查看
 	- https://redis.io/commands/
	- help @sorted_set

zadd key score member		添加一个或多个元素到有序集合, 如果已存在则更新其score值
zrem key member				删除有序集合的一个指定元素
zscore key member			获取有序集合中指定元素的score值
zcard key					获取有序集合中的元素个数
zcount key min max			统计score值在指定范围内的所有元素个数
zincrby key incr member		使有序集合中的指定元素自增, 步长为指定incr值
zrank key member			获取有序集合中指定元素的排名
zrevrank key member			获取有序集合中逆序排序后指定元素的排名
zrange key min max			按照score排序后, 获取指定排名范围内的元素
zrevrange key min max		按照score逆序排序后, 获取指定排名范围内的元素
zrangebyscore key min max	按照score排序后, 获取指定score范围内的元素
zrevrangebyscore key min max	按照score逆序排序后, 获取指定score范围内的元素
zinter|zunion|zdiff			求交|并|差集

score默认为正序, 正序排序为从小到大, 逆序为从大到小, 一般是在命令中加rev

使用Redis处理方案

短信登陆: Redis的共享session应用
商户查询缓存: 企业缓存使用技巧、缓存雪崩穿透等问题
达人探店: 基于List的点赞列表, 基于SortedSet的点赞排行榜
优惠券秒杀: Redis计数器、Lua脚本Redis、分布式锁、三种消息队列
好友关注: 基于Set集合的关注、取关、共同关注、消息推送等功能
附近的商户: Redis的GeoHash应用
用户签到: BitMap数据统计功能
UV统计: HyperLogLog统计功能

短信登陆

基于session实现登陆流程

**集群的session共享问题:**多台服务器不共享session存储空间, 当请求被切换到不同服务器时导致数据丢失

**替代方案特征:**数据共享, 内存存储, key-value结构

基于Redis实现共享session登陆流程

需要考虑问题: 
	选择合适的数据结构		String(json) 或 Hash(对象)
	选择合适的key		使用login:user:[手机号]存储验证码, 使用login:token:[随机数]存储用户信息
	选择合适的存储粒度		仅存储非敏感用户信息

商户缓存查询

缓存

数据交换的缓冲区(cache), 是存贮数据的临时地址, 一般读写性能较高, 可以使用Redis充当应用层缓存

优势:
 - 减低后端负载
 - 提高读写效率, 降低响应时间
劣势:
 - 数据一致性成本
 - 代码维护成本
 - 运维成本(集群, 预热)

添加Redis缓存

缓存更新策略

1
2
3

业务场景:
 - 低一致性场景: 使用内存淘汰机制, 如店铺类型等几乎不变动的查询缓存
 - 高一致性场景: 主动更新为主, 并以超时剔除作为兜底, 如店铺详情查询缓存

一般由缓存调用者在更新数据库时更新缓存, 需要考虑如下问题:

删除缓存/更新缓存
- 更新缓存可能造成多次无效写操作, 推荐使用删除缓存, 等查询时再更新缓存
保证缓存与数据库操作同时成功/失败
- 单体系统可以使用事务保证原子性, 分布式考虑利用TCC等分布式事务方案
先操作缓存/先操作数据库
- 推荐先操作数据库, 再删除缓存, 降低线程安全问题发生的概率

缓存穿透

客户端请求的数据在缓存中和数据库中都不存在, 导致每次请求都直接访问数据库, 缓存永远不生效

解决方案:

缓存空对象

将不存在的数据结果也缓存在Redis中, 赋值为null

优点:
 - 实现简单, 维护方便
缺点:
 - 额外的内存消耗
 - 可能造成短期不一致(设置短TTL 或 新增数据时更新缓存)

布隆过滤

在请求缓存时, 先通过布隆过滤器, 若数据库不存在该数据, 则拒绝请求. 若存在, 才会通过

布隆过滤器实际上是一个很长的二进制向量和一系列随机映射hash函数。将数据通过hash运算后映射到二进制向量位上, 当请求参数时, 同样运算后判断结果位是否都为1, 便可以判断是否存在该数据.

Redis中支持 setbit 和 getbit 操作, 可以利用其实现布隆过滤操作

优点:
 - 内存占用少, 没有多余的key
缺点
 - 实现复杂
 - 存在误判可能(不存在数据精确判断, 存在数据可能误判)

其他预防方式

增强ID复杂度, 避免被猜测ID规律
做好数据的基础格式校验
加强用户权限校验
做好热点参数限流, 峰值限流

缓存雪崩

同一时段内大量缓存key失效或Redis服务宕机, 导致大量请求到达数据库, 带来数据库压力

解决方案

- 给不同key的TTL添加随机值
- 利用Redis集群提高服务可用性(主-从, 哨兵)
- 给缓存业务添加降级限流策略(保护Redis/数据库服务)
- 给业务添加多级缓存

缓存击穿

也叫热点key问题, 就是一个被高并发访问并且缓存重建业务较复杂的key突然失效, 在这期间大量请求直接访问数据库, 带来大量服务压力

解决方案

互斥锁

当发现无缓存数据时, 由当前进程setnx获取互斥锁, 然后进行缓存重建, 其余进程无法获取锁, 等待或返回空数据

优点:
 - 没有额外的内存消耗
 - 保证一致性
 - 实现简单
缺点:
 - 线程需要等待, 性能受影响
 - 可能有死锁风险

使用互斥锁方式解决缓存击穿问题逻辑:

逻辑过期

缓存数据永不过期, 额外添加过期字段, 当用户请求发现该字段已过期时, 加锁并重建缓存, 其余进程无法获取锁, 返回该已过期数据

优点:
 - 线程无需等待, 性能较好
缺点:
 - 不保证一致性
 - 有额外的内存消耗
 - 实现复杂

使用逻辑过期方式解决缓存击穿问题逻辑:

缓存封装工具

方法1: 将任意对象序列化为json并存储在String类型的key中, 并可以设置TTL过期时间

方法2: 将任意对象序列化为json并存储在String类型的key中, 并可以设置逻辑过期时间, 用于处理缓存击穿

方法3: 根据指定key查询缓存, 并解析为指定类型, 利用缓存空值null的方式解决缓存穿透问题

方法4: 根据指定key查询缓存, 并解析为指定类型, 利用逻辑过期解决缓存击穿问题

优惠券秒杀

全局唯一ID

Redis自增

优惠券-用户订单表中, 不推荐使用自增ID, 因为规律性太明显, 且受表单数据量的限制(分表/分库)

全局ID生成器:
	在分布式系统下生成全局唯一ID的工具
需要具备: 唯一性, 高可用, 高性能, 递增性, 安全性
可以基于Redis的incr特性, 构建ID生成器

组成部分: 
 - 符号位: 1bit, 永远为0
 - 时间戳: 31bit, 以秒为单位, 可以使用69年
 - 序列号: 32bit, 秒内计数器, 支持每秒产生2^32个不同的ID
注意: 
 - key可以使用: incr:[业务类型]:[本日日期(天)]来构建
 - 时间戳拼接时可以使用位左移实现 timestap << 32, 保证拼接结果为数字
 - 结果转为10进制

其他方案

1
2
3

- UUID 
- 雪花算法snowflake 
- 数据库自增 (特定表生成序列号部分)

实现优惠券秒杀下单

超卖问题

超卖问题是典型的多线程安全问题, 常见解决方案就是加锁:

悲观锁:

认为线程安全问题一定会发生, 因此在操作数据前先获取锁, 确保线程串行执行. 如mysql的for update 和 Redis的setnx

实现方式简单粗暴, 但性能一般
乐观锁:

认为线程安全问题不一定发生, 因此不加锁, 只是在更新数据时判断有没有其他线程对数据做了修改, 如果没有修改则认为是安全的, 才更新数据. 如果已经被其他线程修改, 则说明发生了线程安全问题, 尝试重试或异常

优点是性能好, 但存在成功率较低的问题

乐观锁实现方式

版本号法: 添加版本号字段, 修改前与获取版本号对照

CAS - Compare And Set: 对照原有数据, 没有修改才更新数据

为解决失败率过高问题, 可以修改stock = 1 条件为 stock > 0

一人一单

修改秒杀业务, 要求同一个优惠券, 一个用户只能下一单

需要为每个用户的行为加悲观锁, 以用户ID作为锁key, 否则查询订单结果不存在会被多个进程执行

集群服务器中的并发安全问题

锁只能保证在本服务器中线程安全, 集群服务时会失效

分布式锁

满足分布式系统或集群模式下多进程可见并且互斥的锁

实现方式:

	MySQL	Redis	Zookeeper
互斥	利用mysql自身的互斥锁机制	利用setnx互斥命令	利用节点的唯一性和有序性实现互斥
高可用	主从保证好	好	好
高性能	一般	好	一般
安全性	断开连接, 自动释放锁	服务宕机,未释放锁可以利用锁超时时间, 到期释放	临时节点, 断开连接自动释放

基于Redis实现分布式锁

获取锁
 - 互斥: 确保只有一个线程可以获取到锁
	setnx lock thread1
 - 设置过期时间
 	expire lock 10
 - 为保证两条命令的原子性, 可以使用 set lock thread1 EX 10 NX 一次性执行
释放锁
 - 手动释放: del lock
 - 超时释放

非阻塞式获取锁

问题一: 误删他锁

该流程中可能发生误删锁的状况, 流程如下:

解决方案:

需要在加锁时添加自己进程/服务身份的标识(uuid+id), 删除锁前判断该锁是否属于本进程/服务, 再执行释放锁操作

修改流程为:

问题二: 判断/删锁原子性

在判断锁标识是自己的之后, 由于系统阻塞, 阻塞期间锁超时释放, 导致其他锁被误删

解决方案:

Lua脚本, 在一个脚本中编写多条Redis命令, 确保多条命令执行时的原子性

流程:
 - 获取锁的线程/服务标识
 - 判断是否与当前标识一致
 - 如果一致则释放锁(删除)
 - 如果不一致则什么都不做

1	Redis-Lua脚本文档: https://redis.io/docs/manual/programmability/eval-intro/

-- Lua脚本 unlock.lua
-- 获取参数中锁的key
local key = KEYS[1]
-- 当前线程/服务标识
local threadId = ARGV[1]
-- 获取锁中的线程/服务标识
local id = redis.call('get', key)

-- 比较线程/服务标识是否与锁标识一致
if id == threadId then
    return redis.call('del', key)
end
return 0

其他问题

不可重入

同一线程无法多次获取同一把锁( 函数调用 )

可通过使用Redis-hash类型数据, 同时加入锁次数字段.
当同一个线程/服务再次请求时, 为锁字段添加1, 同时添加有效期.
执行业务结束后, 释放锁, 即锁次数减1. 
当所有业务结束, 锁次数为0时, 删除锁对应key

-- 获取锁 tryLock.lua
local key = KEYS[1]				-- 锁的key
local threadId = ARGV[1]		-- 线程/服务唯一标识
local releaseTime = ARGV[2]		-- 锁过期时间

if redis.call('exists', key) == 0 then			-- 判断是否存在锁
    redis.call('hset', key, threadId, 1)		-- 不存在, 获取锁
    redis.call('expire', key, releaseTime)		-- 设置有效期
    return 1									-- 返回成功结果
end
if redis.call('hexists', key, threadId) == 1 then	-- 锁已存在, 判断是否可以重入
    redis.call('hincrby', key, threadId, 1)		-- 获取锁, 重入次数加1
    redis.call('expire', key, releaseTime)			-- 重置锁有效期
    return 1										-- 返回成功结果
end
return 0						-- 获取锁失败

-- 释放锁 unlock.lua
local key = KEYS[1]				-- 锁的key
local threadId = ARGV[1]		-- 线程/服务唯一标识
local releaseTime = ARGV[2]		-- 锁过期时间

if redis.call('hexists', key, threadId) == 0 then		-- 判断当前锁是否被自己持有
    return nil											-- 不是自己持有, 返回
end
local count = redis.call('hincrby', key, threadId, -1)	-- 是自己持有, 次数加1
if count > 0 then										-- 判断重入次数是否已经为0
    redis.call('expire', key, releaseTime)				-- 大于0说明不能释放锁, 重置过期时间
    return nil
else
    redis.call('del', key)								-- 等于0说明可以释放锁, 删除
    return nil
end

不可重试

获取锁只尝试一次就返回false, 没有重试机制

1	利用信号量和Redis-PubSub功能实现等待、唤醒, 获取锁失败的重试机制

超时释放

锁超时释放虽然可以避免死锁, 但在业务耗时较长的场景, 存在安全隐患
1
利用watchDog, 每隔一段时间(releaseTime/3), 重置超时时间
主从一致性

如果在主从集群中, 同步存在延迟, 从服务器还未同步到主服务器的写入的锁数据, 其余进程读从服务器数据, 可能导致出现多个锁
1
multilock, 同时在多个同级节点设置锁, 仅在所有节点可以获取锁才算成功

Redis秒杀优化

可以将判断库存/校验一人一单操作需要信息放入Redis, 再通过Lua脚本执行, 只有校验通过的用户请求才会被后台继续执行, 实际操作数据库, 流程如下:

其中库存数据可以通过String格式存储, 一人一单验证则通过将用户ID放入指定集合, 判断集合是否存在用户判断

加快了处理和响应速度, 同时减轻了数据库的访问压力

实际操作:
 - 新增秒杀优惠券的同时, 将优惠券库存信息存入Redis
 - 基于Lua脚本, 判断秒杀库存、一人一单, 决定用户是否抢购成功
 - 若抢购成功, 将优惠券ID和用户ID封装后存入阻塞队列
 - 开启线程任务, 不断从阻塞队列中获取信息, 实现异步下单功能

-- 秒杀预处理脚本  seckill.lua
local voucherId = ARGV[1]	-- 优惠券ID, 作为键名
local userId = ARGV[2]		-- 用户ID

local stockKey = 'seckill:stock:' .. voucherId	-- 库存数量Key
local orderKey = 'seckill:order:' .. voucherId	-- 订单用户集合Key

if tonumber(redis.call('get', stockKey)) <= 0 then		-- 若库存数量不足
    return 1	
end
if redis.call('sismember', orderKey, userId) == 1 then	-- 若用户已在集合中
    return 2	-- 说明已经下过单
end

redis.call('decr', stockKey)			-- 扣除Redis缓存中的库存
redis.call('sadd', orderKey, userId)	-- 保存用户到订单集合中
return 0

但该方案仍旧存在内存限制(服务队列消耗本机内存)和数据安全(异常/宕机)问题

Redis消息队列实现异步秒杀

消息队列

消息队列(Message Queue), 即存放消息的队列, 包括三个角色:
 - 消息队列: 存储和管理消息, 也称消息代理(Message Broker)
 - 生产者: 发送消息到消息队列
 - 消费者: 从消息队列中获取消息并处理消息

基于Redis实现消息队列

Redis中提供了三种不同的方式实现消息队列

List结构: 基于列表结构模拟消息队列
PubSub: 基本的点对点消息模型
Stream: 比较完善的消息队列模型

基于List模拟消息队列

优点:
 - 基于Redis存储, 不受限于服务所在内存上限
 - 基于Redis的持久化机制, 数据安全性有保证
 - 满足消息有序性
缺点:
 - 无法避免消息丢失(获取Redis信息后服务宕机)
 - 只支持单消费者

基于PubSub模拟消息队列

1	文档: https://redis.io/commands/?group=pubsub

发布订阅(PubSub)是Redis2.0引入的消息传递模型. 即消费者可以订阅一个或多个频道(channel), 生产者向对应channel发送消息后, 所有订阅者都能收到相关消息. 主要使用以下命令实现

publish channel message			向一个频道发送消息
subscribe channel [channel]		订阅一个或多个频道
psubscribe pattern [pattern]	订阅符合pattern格式的所有频道
	- h?llo		匹配hello,hallo和hxllo
	- h*llo		匹配hllo和heeeello
	- h[ae]llo	匹配hello但hallo,不匹配hillo

优点:
 - 支持多生产、多消费
缺点:
 - 不支持数据持久化
 - 无法避免消息丢失
 - 消息堆积有上限, 超出时数据丢失 (消费者处理缓慢造成堆积)

基于Stream的消息队列

Stream

Stream是Redis5.0引入的一种数据类型, 可以实现一个功能完善的消息队列

1
2
3

类型文档: 
http://www.redis.cn/topics/streams-intro.html
https://redis.io/docs/manual/data-types/streams/

核心命令:

发送消息命令 xadd
添加一个新的entry到stream
- 创建名为users的队列, 并向其中发送一个消息, 内容为{name=Tom, age=5} ,无最大消息数量限制, 且使用Redis自动生成ID
读取消息命令 xread

从一个或者多个流中读取数据，仅返回ID大于调用者报告的最后接收ID的条目。此命令有一个阻塞选项
1
2
3
4
5
命令特点:
- 消息可回溯
- 一个消息队列可以被多个消费者读取
- 可以阻塞读取
- 有消息漏读风险
- 使用xread读取第一条信息
- 阻塞方式读取当前信息

命令效果

使用 xread ... 0 从头获取消息时, 每个线程都可以获取到该队列第一条数据

使用 xread ... block ... $ 阻塞获取最新一条信息

实际使用

可以通过循环调用xread以阻塞方式查询最新消息, 从而实现持续监听队列效果

while (true){
    // 尝试读取队列消息, 最多阻塞2秒
    msg = redis.eval('xread count 1 block 2000 streams users $');
    if (msg == null){
        continue;
    }
    handleMessage(msg)	// 处理消息
}

但当指定起始ID为$时, 代表读取最新消息, 若在处理一条信息时, 有超过1条消息到达队列, 则下次获取时也只能获取到最新一条, 出现 消息漏读 问题

消费者组

Consumer Group: 将多个消费者划分到一个组中, 监听同一个队列.

特征

- 消息分流
	列中的消息会分流给组内的不同消费者，而不是重复消费，从而加快消息处理的速度
- 消息标示
	消费者组会维护一个标示，记录最后一个被处理的消息，哪怕消费者宕机重启，还会从标示之后读取消息。确保每一个消息都会被消费
- 消息确认
	消费者获取消息后，消息处于pending状态，并存入一个pending-list。当处理完成后需要通过XACK来确认消息，标记消息为已处理，才会从pending-list移除。

命令

xgroup CREATE key groupname ID [MKSTREAM]	管理消费者组
 - key 			队列key名称
 - groupname 	消费者组名称
 - ID 			起始ID标示, $代表队列中最后一个消息, 0代表第一个消息
 - MKSTREAM 	队列不存在时自动创建

其他命令:
xgroup DESTORY key groupname	删除指定消费者组
xgroup CREATECONSUMER key groupname cousumername  	给指定的消费者组添加消费者
xgroup DELCONSUMER key groupname consumername		删除消费者组中的指定消费者

从消费者组读取消息:
xreadgroup GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]	
 - group		消费组名称
 - consumer		消费者名称，如果消费者不存在，会自动创建一个消费者
 - count		本次查询的最大数量
 - BLOCK milliseconds	阻塞读取, 当没有消息时最长等待时间
 - NOACK		无需手动ACK，获取到消息后自动确认
 - STREAMS key	指定队列名称
 - ID			获取消息的起始ID：
   - >		从下一个未消费的消息开始
   - 其它    根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从pending-list中的第一个消息开始

确认消息:
xack group ID [ID ...]  	将pending的消息标记为已正确处理，从而有效地将其从使用者组的pending-list中删除.

查看pending-list:
xpending key group start end count  	从消费者组待处理条目列表中返回信息和条目,即获取但从未确认的消息。

命令效果

添加消费者组队列, 并使用消费者组消费Stream数据

查看pending-list数据, 并重新获取数据处理, 确认消息后再查看

获取数据时, 将ID由 > 改为 0, 即可从获取消息转为获取 pending-list 未确认消息

实际使用

while (true){
    //监听队列, 阻塞模式, 最长等待2秒
    msg = redis.eval('xreadgroup GROUP group_x1 consumer1 COUNT 1 BLOCK 2000 STREAMS x1 >');
    if (msg == null){	// 返回结果为null, 说明没有消息, 继续下次循环
        continue;
    }
    try{
        handleMessage(msg);	// 处理消息函数, 内部完成后应使用xack确认
    }catch(Exception e){ 	// 捕获异常, 说明没有完成消息处理
        while (true){
            // 重新获取pending-list中最新消息
            msg = redis.eval('xreadgroup GROUP group_x1 consumer1 COUNT 1 STREAMS x1 0');
            if (msg == null){	// 返回null说明没有异常未确认消息, 结束循环
                break;	
            }
            try{
                handleMessage(msg);	// 返回数据包含未确认异常消息, 再次处理, 完成后xack确认
            }catch(Exception e){
                LogErrors(e)	// 捕获异常, 处理失败, 记录日志, 重新尝试
                continue;		// 或者返回ack, break中断尝试
            }
        }
    }
}

特征

- 消息可回溯
- 可以多消费者争抢消息, 加快消费速度
- 可以阻塞读取
- 没有消息漏读的风险
- 有消息确认机制, 保证消息至少被消费一次

三种模式对比

	List	PubSub	Stream
消息持久化	支持	不支持	支持
阻塞读取	支持	支持	支持
消息堆积处理	受限于内存空间大小, 可以利用多消费者加快处理	受限于消费者缓冲区	受限于队列长度, 可以利用消费者组提高消费速度, 减少堆积
消息确认机制	不支持	不支持	支持
消息回溯	不支持	不支持	支持

基于Stream实现异步秒杀下单

创建一个Stream类型的消息队列，名为stream.orders
1
xgroup CREATE stream.order group1 0 MKSTREAM

修改秒杀下单Lua脚本，在认定有抢购资格后，直接向stream.orders中添加消息，内容包含voucherId、userId、orderId

-- 秒杀预处理脚本  seckill.lua
local voucherId = ARGV[1]	-- 优惠券ID, 作为键名
local userId = ARGV[2]		-- 用户ID
local orderId = ARGV[3]		-- 订单ID

local stockKey = 'seckill:stock:' .. voucherId	-- 库存数量Key
local orderKey = 'seckill:order:' .. voucherId	-- 订单用户集合Key

if tonumber(redis.call('get', stockKey)) <= 0 then		-- 若库存数量不足
    return 1	
end
if redis.call('sismember', orderKey, userId) == 1 then	-- 若用户已在集合中
    return 2	-- 说明已经下过单
end

redis.call('decr', stockKey)			-- 扣除Redis缓存中的库存
redis.call('sadd', orderKey, userId)	-- 保存用户到订单集合中

-- 发送消息到stream队列中
redis.call('xadd', 'stream.order', '*', 'userId', userId, 'voucherId', voucherId, 'orderId', orderId)
return 0

服务代码

1	redis.eval('path_of_seckill_lua_script',[voucherId, userId, orderId])

项目启动时，开启任务，尝试获取stream.orders中的消息，完成下单

while (true){
    try{
        // 获取消息队列中的订单信息
        xreadgroup GROUP group1 consumer1 COUNT 1 BLOCK 2000 STREAMS stream.order >
        // 判断消息获取是否成功
            // 若获取失败, 说明没有消息, 继续下次循环
        // 若获取成功, 可以下单
        // ACK确认
        xack stream.order group1 [ID]
    }catch(Exception e){
        // 订单处理异常
        while(true){
            try{
                // 获取pending-list队列中的未确认订单信息
                xreadgroup GROUP group1 consumer1 COUNT 1 STREAMS stream.order 0
                // 判断消息获取是否成功
                    // 若获取失败, 说明pending-list没有消息, break
                // 若获取成功, 可以下单
                // ACK确认
            }catch(Exception e){
                // 处理pending-list异常信息 sleep + continue + count + break / log + break
            }
        }
    }
}

达人探店

发布探店笔记

使用Redis中的Set数据结构, 实现点赞功能

1
2
3

- 将当前blog对应的点赞人员ID放入以blogId为键的Set集合中. 
  用sismember判断是否点赞, 若当前用户存在于集合中, 则已点赞, 若不存在, 说明未点赞
- 修改点赞功能, 点赞过用srem取出该用户ID, 数据库点赞数减一, 未点赞用sadd加入该用户ID, 数据库点赞数加一

点赞排行榜

使用Redis中的SortedSet数据结构, 实现点赞排行功能

在blog详情页, 需要添加点赞列, 显示前几个点赞用户头像
- 将点赞用户ID放入SortedSet集合, 以点赞动作时间戳作为score
- 查询用户ID是否存在时, 可以用zscore查询对应元素是否存在
- 选出前几个点赞用户ID时, 可以用zrange依据时间戳排序获取

好友关注

关注和取关

1 2	- 在关注时, 使用sadd将目标用户ID放入以用户ID为key的set集合中 - 取关时, 使用srem移除用户ID

共同关注

使用Redis中的Set数据结构, 实现显示当前用户与指定用户的共同好友

1	- 共同关注, 获取当前用户与目标用户的set集合,使用sinter获取交集,即为共同关注用户ID集合

关注推送

也叫Feed流. 为用户持续的提供, 通过无限下拉刷新获取新的信息, 有两种常见模式:

TimeLine:

不做内容筛选, 简单按照内容发布时间排序, 常用于好友或关注信息. 如朋友圈
智能排序:

利用算法屏蔽违规、不感兴趣内容. 推送感兴趣的信息吸引用户

TimeLine模式实现方式:

拉模式

也叫读扩散, 只有在关注后查看, 才会主动从目标用户的发件箱中获取信息
推模式

也叫写扩散, 发送消息时向所有关注者同时发送信息
推拉结合

也叫读写混合, 兼具推和拉模式的优点

	拉模式	推模式	推拉结合
写消耗	低	高	中
读消耗	高	低	中
用户读取延迟	高	低	低
实现难度	简单	简单	复杂
使用场景	很少使用	用户量少, 没有大V	千万级别用户量,有大V

Feed流中的滚动分页

由于Feed流中的数据会不断更新, 数据角标索引也在实时变化, 所以不能采用传统分页模式, 而是使用SortedSet数据结构, 使用时间戳为score, 查询分页时记录lastId, 下一页数据从lastId开始查询

命令参数:
zrevrangebyscore key max min [WITHSCORES] [LIMIT offset count]
 - max: 最大分数, 可以使用 +inf 代表最大值
 - min: 最小分数, 可以使用 -inf 代表最小值
 - offset: 偏移量, 指定score后继续向后偏移取值
 - count: 返回取值的数量

使用推模式完成推送

- 被关注用户发送blog时, 查询所有关注者, 同时使用Redis中的SortedSet数据结构记录blog_ID到关注者ID的key, 作为关注者收件箱内容, score为发布时间戳
- 关注者收件箱中存在SortedSet数据结构, 存储多位关注者锁发blogId, 前端发送上一次查询的最小时间戳lastId(初始值为当前时间戳, 后续从后端获取)和偏移量offset(初始值为0, 后续由后端返回偏移量取值)
- 获取第一页数据时, 后端获取用户当前时间戳作为lastId, 偏移量offset, 执行命令:
	zrevrangebyscore key lastId -inf WITHSCORES limit offset 3
	并返回最后一个ID和最后一个元素score值重复出现的次数作为下次请求offset
- 获取后续页数据时, 后端获取lastId和offset, 执行命令
	zrevrangebyscore key lastId -inf WITHSCORES limit offset 3
	并返回最后一个ID和最后一个元素score值重复出现的次数作为下次请求offset

简便方式: 在添加score时, 在时间戳后添加随机数或有序自增数

附近商户

GEO数据结构

地理坐标Geolocation, Redis3.2版本后提供, 允许存储地理经纬度信息, 常见命令有:

geoadd 		添加一个地理空间信息到SortedSet,经纬度作为score,名称作为member
geodist 	计算指定两个点或两个member之间的距离
geohash		将指定member坐标转为hash字符串并返回
geopos		返回指定member的坐标
georadius	以指定圆点, 半径, 返回圆范围内的所有member
geosearch	在指定范围内搜索member, 可以指定圆形还是矩形, 6.2版本后提供
geosearchstore	功能与geosearch一致, 但可以将结果存入一个指定的key, 6.2版本后提供

附近商户

- 将店铺所在经纬度x,y存入Redis的GEO数据类型中, 可以考虑以搜索类型分类分Key存储
  - 查询店铺信息
  - 将店铺按照TypeId分组, typeId一致的放入同一个GEO集合
  - 分批完成写入Redis:
  	geoadd shop:geo:[typeId] [LON] [LAT] [shopId] ...
- 获取用户当前经纬度x,y. 执行命令:
  geosearch shop:geo:[typeId] FROMLONLAT x y BYRADIUS 10 km COUNT 100 WITHDIST
- 截取分页数据后获取商户ID集合[key=shopId,value=distance]
- 有序查询出商户具体数据
  select * from table where id in [shopIds] order by field(id,[idsString])
- 整理后返回前端

用户签到

BitMap数据结构

Redis中是利用string类型数据结构实现BitMap，因此最大上限是512M，转换为bit则是 2^32个bit位

SETBIT		向指定位置（offset）存入一个0或1
GETBIT		获取指定位置（offset）的bit值
BITCOUNT	统计BitMap中值为1的bit位的数量
BITFIELD 	操作（查询、修改、自增）BitMap中bit数组中的指定位置（offset）的值
BITFIELD_RO	获取BitMap中bit数组，并以十进制形式返回
BITOP		将多个BitMap的结果做位运算（与 、或、异或）
BITPOS		查找bit数组中指定范围内第一个0或1出现的位置

签到功能

如果使用数据库实现功能, 可能造成数据大表出现

可以按月统计用户签到数据, 签到记录为1, 未签到记录为0

把每一个bit位对应当前的每一天, 形成映射关系. 用0和1标识业务状态, 即位图(BitMap)

- 获取当前用户
- 获取当前日期
- 依据用户ID和日期生成当月key:sign:[uid]:202203
- 根据当前月天数写入BitMap
  setbit key offset 1

签到统计

本月已连续签到天数

从最后一次签到开始向前统计，直到遇到第一次未签到为止，计算总的签到次数，就是连续签到天数。

获取一个月全部签到数据
1
bitfield key get u[天数] 0

从后向前遍历每个bit位

1	整体向右移动一位, 依次与1做与运算获取+最后一个bit位

实现

- 获取本月截止当前天的所有签到记录
  bitfield sign:[uid]:202203 GET u[天数] 0
- 循环遍历while
  - 让结果与1做与运算, 获取最后一个Bit位, 若为0, 结束循环, 若为1, 计数器加1
  - 右移一位 (num >> 1), 继续下次循环
- 返回连续签到次数计数器

UV统计

UV(Unique Visitor), 也叫独立访客量，是指通过互联网访问、浏览这个网页的自然人。1天内同一个用户多次访问该网站，只记录1次。

PV(Page View), 也叫页面访问量或点击量，用户每访问网站的一个页面，记录1次PV，用户多次打开页面，则记录多次PV。往往用来衡量网站的流量。

HyperLogLog数据结构

Hyperloglog(HLL)是从Loglog算法派生的概率算法，用于确定非常大的集合的基数，而不需要存储其所有值。相关算法原理可以参考：HyperLogLog 算法的原理讲解以及 Redis 是如何应用它的

Redis中的HLL是基于string结构实现的，单个HLL的内存永远小于16kb。作为代价，其测量结果是概率性的，有小于0.81％的误差。不过对于UV统计来说，这完全可以忽略。

实现UV统计

1 2	- 用户访问时, 将用户信息存入Redis中的HyperLogLog数据结构中 - 统计时, 可以使用pfmerge将每月/每年数据合并, 通过频繁count获取UV数据

PS:

1	课程文档: https://pan.baidu.com/s/1189u6u4icQYHg_9_7ovWmA?pwd=eh11

Redis结构功能解析

简介

NoSQL

特征

安装

apt安装

编译安装

配置

运行

关闭

配置systemctl

通过systemctl配置开机自启动

客户端

命令行客户端

图形化桌面客户端

数据结构

通用命令

String类型命令

Hash类型命令

List类型命令

Set类型命令

SortedSet类型命令

使用Redis处理方案

短信登陆

基于session实现登陆流程

基于Redis实现共享session登陆流程

商户缓存查询

缓存

添加Redis缓存

缓存更新策略

缓存穿透

缓存空对象

布隆过滤

其他预防方式

缓存雪崩

解决方案

缓存击穿

互斥锁

逻辑过期

缓存封装工具

优惠券秒杀

全局唯一ID

实现优惠券秒杀下单

超卖问题

一人一单

分布式锁

基于Redis实现分布式锁

非阻塞式获取锁

问题一: 误删他锁

**问题二: 判断/删锁 原子性 **

其他问题

Redis秒杀优化

Redis消息队列实现异步秒杀

消息队列

基于Redis实现消息队列

基于List模拟消息队列

基于PubSub模拟消息队列

基于Stream的消息队列

Stream

消费者组

三种模式对比

基于Stream实现异步秒杀下单

达人探店

发布探店笔记

点赞

点赞排行榜

好友关注

关注和取关

共同关注

关注推送

附近商户

GEO数据结构

附近商户

用户签到

BitMap数据结构

签到功能

签到统计

UV统计

HyperLogLog数据结构

实现UV统计

问题二: 判断/删锁原子性