Redis 学习笔记

参考文档

• Redis 命令官方文档
• Redis 中文命令文档

Redis 介绍

为什么需要 Redis

关系型数据库 MySQL 的性能瓶颈主要在于磁盘 I/O 性能较低，数据关系复杂时扩展性较差，不便于构建大规模集群。

Redis 基于内存存储，可以减少 I/O 次数。它没有关系型数据库的表结构，也不使用 SQL。

Redis 特点

Redis 是基于内存的键值对存储系统。key 是字符串，value 支持多种数据类型。

Redis 安装与配置

在 Linux 系统安装

安装编译工具并解压源码：

1
2
3
4
5
yum install gcc
cd /opt
tar zxvf redis-6.2.6.tar.gz
cd /opt/redis-6.2.6
make install

复制配置文件，启用后台运行后启动 Redis：

1
2
3
4
cp /opt/redis-6.2.6/redis.conf /etc/
vi /etc/redis.conf
# 将 daemonize 修改为 yes
redis-server /etc/redis.conf

常用工具：

• redis-benchmark：性能测试工具。
• redis-check-aof：修复有问题的 AOF 文件。
• redis-check-dump：修复有问题的 dump.rdb 文件。
• redis-sentinel：启动 Redis Sentinel。
• redis-server：启动 Redis 服务端。
• redis-cli：Redis 命令行客户端。

连接和关闭服务：

1
2
3
4
5
6
7
# 连接本机默认端口 6379
redis-cli
redis-cli -p 6379

# 关闭 Redis
redis-cli shutdown
redis-cli -p 6379 shutdown

/usr/local/bin 通常已在环境变量中，因此可以直接执行上述命令。

修改 Redis Server 端口

编辑 redis.conf：

1
port 6379

Redis 指令

Redis CLI 基础指令

Key 操作

数据库操作

Redis 默认提供 16 个逻辑库（0-15），启动后默认使用 0 号库，各库数据相互隔离。

Redis 数据类型

数据存储格式

• Redis 本身可以理解为一个 Map，所有数据都以 key-value 形式存储。
• key 是字符串，value 支持多种数据类型和结构。

String

String 是 Redis 最基本的数据类型，一个 key 对应一个 value。它是二进制安全的，可以保存文本、图片或序列化对象，单个 value 最大为 512 MB。

1
2
3
4
5
6
7
SET key "hello world"
GETRANGE key 0 4
# "hello"

SETRANGE key 6 redis
GET key
# "hello redis"

List

List 用于保存多个有序元素，元素可以重复。底层采用双向链表，两端操作性能较高，按索引访问中间节点的性能相对较低。

1
2
3
4
5
Node {
    Node pre
    Node next
    T data
}

List 适合保存有先后顺序的数据，例如通知流。作为队列时可使用 RPUSH + LPOP，作为栈时可使用 LPUSH + LPOP。

Set

Set 是无序且元素唯一的集合，底层使用字典/哈希表实现。

Hash

Hash 是 field-value 键值对集合，适合存储对象，类似 Java 中的 Map<String, Object>。

1
2
HSET user:1001 id 1001 name zangxin age 20 gender male
HGET user:1001 name

Sorted Set / ZSet

ZSet 与 Set 类似，成员不能重复；区别在于每个成员都关联一个可重复的分数 score，集合会按分数排序。

1
2
3
ZADD dota2_hero 1 dushe 2 feiji 3 xiaoniu 4 xiaohei 5 zeous
ZRANGE dota2_hero 0 -1 WITHSCORES
ZSCORE dota2_hero feiji

Redis 配置

参考资料

• https://www.cnblogs.com/nhdlb/p/14048083.html#_label0

常规配置

日志级别

将日志写入文件时，可以配置：

1
logfile /var/log/redis/redis.log

需要确保目录存在，并且 Redis 进程拥有写入权限。

密码管理

Units 单位

1
2
3
4
5
6
# 1k  => 1000 bytes
# 1kb => 1024 bytes
# 1m  => 1000000 bytes
# 1mb => 1024*1024 bytes
# 1g  => 1000000000 bytes
# 1gb => 1024*1024*1024 bytes

配置大小单位,开头定义了一些基本的度量单位，只支持 bytes，不支持 bit 不区分大小写

Include

• 多实例的情况可以把公用的配置文件提取出来, 然后 include

1
2
# include /path/to/local.conf
# include /path/to/other.conf

网络

远程访问不应只依靠注释 bind 或关闭保护模式。生产环境还应配置密码或 ACL，并通过防火墙、安全组限制访问来源。

为什么需要 TCP Keepalive

短连接完成数据交互后会主动释放连接。长连接可能因客户端断网、进程异常等原因未能正常关闭，服务端也无法立即感知，长期积累会形成大量半打开连接。

Keepalive 可以帮助服务端更快发现无效连接，减少系统资源浪费，也避免继续向已经失效的连接发送业务数据。

限制

客户端连接数

1
maxclients 10000

最大内存

maxmemory 用于限制 Redis 可使用的最大内存：

1
maxmemory <bytes>

• 32 位 Redis 实例通常最多使用约 3 GB 内存。
• 64 位实例默认没有该限制，但生产环境仍建议显式设置上限。
• 实际值取决于数据规模、持久化方式以及同机运行的其他服务，不能使用一个固定值套用所有环境。

核心原则是从保守值开始测试，通过实际监控调整，确保 Redis 扩容不会影响系统中其他程序的正常运行。

内存淘汰策略

maxmemory-policy 决定 Redis 达到内存上限后如何处理新的写入请求。

noeviction 是传统默认策略。纯缓存场景通常更关注 allkeys-lru 等可淘汰所有 key 的策略，具体选择应根据数据是否允许丢失决定。

淘汰采样数量

LRU 和最小 TTL 淘汰使用近似算法。maxmemory-samples 控制每次选择候选 key 时的采样数量：

1
maxmemory-samples 5

通常可以设置为 3～7。采样数量越大，结果越接近理想算法，但 CPU 开销也会增加。

发布与订阅 / 消息队列

Redis 发布订阅（Pub/Sub）是一种消息通信模式：发布者发送消息，订阅者接收消息。

发布者向频道发送消息后，当前正在订阅该频道的客户端会立即收到消息。

消息队列模型

从机制上看，Pub/Sub 更接近广播系统。它与可靠任务队列不同：消息不会持久化，也不会为离线消费者保留。

发布订阅模式分类

命令行实现发布和订阅

注意

• 发布的消息没有持久化
• 订阅的客户端, 只能收到订阅后发布的消息

Jedis

Jedis 是 Java 程序操作 Redis 的客户端工具。

文档

• Jedis 官方文档

Maven 依赖

1
2
3
4
5
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>5.2.0</version>
</dependency>

允许 Redis 远程连接

• 注释掉 bind 配置。
• 将保护模式设置为 protected-mode no。
• 开放 Linux 的 6379 端口。

连接

1
2
3
4
5
6
7
// 连接redis
Jedis jedis = new Jedis("192.168.2.85", 6379);
jedis.auth("root");
String res = jedis.ping();
System.out.println("res = " + res); // pong
// 关闭连接
jedis.close();

操作 Key

1
2
3
4
5
6
7
8
9
jedis.set("k1", "v1");
jedis.set("k2", "v2");
jedis.set("k3", "v3");
Set<String> keys = jedis.keys("*");
keys.forEach(System.out::println);

System.out.println(jedis.exists("k1"));// true
System.out.println(jedis.ttl("k2"));// -1
System.out.println(jedis.get("k3")); // v3

操作 String

1
2
3
jedis.mset("k1", "vv1", "k2", "vv2", "k3", "vv3","k4", "vv4");
List<String> stringKeyList = jedis.mget("k1", "k2", "k3", "k4");
stringKeyList.forEach(System.out::println);

操作 List

1
2
3
4
5
// 添加数据
jedis.lpush("name_list", "tom", "jerry", "zangxin", "alice");
// 取出数据
List<String> nameList = jedis.lrange("name_list", 0, -1);
nameList.forEach(System.out::println);

操作 Set

1
2
3
jedis.sadd("heroes","dushe","xiaohei","feiji","zeous","luna");
Set<String> heroes = jedis.smembers("heroes");
heroes.forEach(System.out::println);

操作 Hash

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
jedis.hset("zangxin", "name", "zangxin");
jedis.hset("zangxin", "age", "20");
jedis.hset("zangxin", "gender", "male");
jedis.hset("zangxin", "game", "dota2");

String game = jedis.hget("zangxin", "game");
System.out.println(game);

// 第二种放入方式-->map
HashMap<String, String> map = new HashMap<>();
map.put("name", "zangxin");
map.put("age", "22");
jedis.hset("zangxin2", map);

List<String> values = jedis.hmget("zangxin", "name", "age", "gender", "game");
values.forEach(System.out::println);

操作 ZSet

1
2
3
4
5
6
7
8
9
10
jedis.zadd("role", 1, "桐人");
jedis.zadd("role", 3, "亚丝娜");
jedis.zadd("role", -1, "诗乃");
jedis.zadd("role", 2, "莉法");

List<String> role = jedis.zrange("role", 0, -1);
role.forEach(System.out::println);// 诗乃,桐人,莉法,亚丝娜
// 逆序
List<String> reverse = jedis.zrevrange("role", 0, -1);
reverse.forEach(System.out::println);// 亚丝娜,莉法,桐人,诗乃

Spring Boot 整合 Redis

相关内容见 Spring Boot 学习笔记。

Redis 持久化

Redis 持久化是将内存中的数据写入磁盘。

持久化策略

RDB（Redis Database）

RDB 会在指定时间将内存数据生成快照并写入磁盘，恢复时再将快照文件加载到内存。

执行流程

1. 客户端执行 BGSAVE，或由配置自动触发。
2. 主进程检查是否已有持久化子进程在运行。
3. 如果没有，主进程通过 fork() 创建子进程。fork() 期间会短暂阻塞，随后主进程继续处理请求。
4. 子进程借助 Copy-On-Write 将数据写入临时 RDB 文件。
5. 写入完成后，临时文件原子替换旧的 dump.rdb，并通知主进程。

参考：Linux fork 与 Copy-On-Write。

常用配置

1
2
3
4
5
6
7
8
9
10
dbfilename dump.rdb
dir ./

save 3600 1
save 300 100
save 60 10000

stop-writes-on-bgsave-error yes
rdbcompression yes
rdbchecksum yes

• dbfilename：RDB 文件名。
• dir：RDB 文件保存目录，可以配置绝对路径。
• save 60 10000：60 秒内至少有 10000 个 key 变化时创建快照。
• stop-writes-on-bgsave-error：快照写入失败时是否停止写操作。
• rdbcompression：是否使用 LZF 压缩快照。
• rdbchecksum：是否使用 CRC64 校验文件完整性。

正常关闭时，SHUTDOWN 或 SHUTDOWN SAVE 会保存 RDB；SHUTDOWN NOSAVE 不保存。

手动操作

恢复数据时，将备份文件复制为 dump.rdb 并放入配置目录，然后重启 Redis。

优点

• 文件紧凑，节省磁盘空间。
• 适合大规模数据恢复，恢复速度通常快于 AOF。
• 主进程不直接执行磁盘 I/O，运行性能较好。

缺点

• 异常宕机时可能丢失最后一次快照之后的数据。
• 数据量较大时，fork() 和 Copy-On-Write 会消耗额外资源。

AOF（Append Only File）

AOF 以日志形式追加记录写命令，例如 SET 和 DEL，不会记录 GET 等读命令。Redis 重启时会重新执行 AOF 中的命令来恢复数据。

执行流程

1. 写命令追加到 AOF 缓冲区。
2. 根据 appendfsync 策略将缓冲区同步到磁盘。
3. 文件达到重写条件时执行 Rewrite，去除冗余命令并压缩文件。
4. Redis 重启时加载 AOF 文件并重建数据。

开启 AOF

1
2
appendonly yes
appendfilename "appendonly.aof"

AOF 与 RDB 同时开启时，Redis 通常优先使用 AOF 恢复数据。

同步策略

恢复与修复

1. 备份损坏的 AOF 文件。
2. 执行 /usr/local/bin/redis-check-aof --fix appendonly.aof。
3. 重启 Redis 并重新加载 AOF。

修复可能造成部分数据丢失，因此应先保留原文件。

Rewrite

Rewrite 会删除无效的历史命令，并将多条可合并命令压缩为更少的命令。可以手动执行 BGREWRITEAOF，也可以自动触发：

1
2
auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb

• auto-aof-rewrite-min-size：AOF 文件达到该大小后才允许自动重写。
• auto-aof-rewrite-percentage：当前文件相对上次重写后的增长比例达到该值时触发重写。

优点

• 备份机制更稳健，丢失数据的概率更低。
• 日志可读，便于排查和修复误操作。

缺点

• 占用磁盘空间通常大于 RDB。
• 恢复速度通常慢于 RDB。
• 频繁同步会带来额外性能开销。

其他策略

• No persistence：不启用持久化，适合纯缓存场景。
• RDB + AOF：同时启用两种机制，兼顾恢复速度与数据完整性。

选择持久化策略

参考 Redis 持久化官方文档。

官方不建议单独使用 AOF。定期创建 RDB 快照有利于数据库备份、快速重启，也能在 AOF 引擎出现问题时提供恢复手段。

Redis 事务与超卖问题

定义

• 1、Redis 事务是一个单独的隔离操作:事务中的所有命令都会序列化、按顺序地执行
• 2、事务在执行的过程中，不会被其他客户端发送来的命令请求所打断
• 3、Redis 事务的主要作用就是串联多个命令防止别的命令插队

Redis 事务的三个特性

事务指令

案例/注意点

需求：使用 Redis 事务依次添加 tx1-x1、tx2-x2、tx3-x3。

全部成功

1
2
3
4
5
6
7
8
9
10
11
12
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set  tx1 x1
QUEUED
127.0.0.1:6379(TX)> set tx2 x2
QUEUED
127.0.0.1:6379(TX)> set tx3 x3
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
2) OK
3) OK

放弃事务

1
2
3
4
5
6
7
8
9
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set tx1 x1
QUEUED
127.0.0.1:6379(TX)> set tx2 x2
QUEUED
127.0.0.1:6379(TX)> DISCARD
OK
127.0.0.1:6379>

组队阶段发生错误

1
2
3
4
5
6
7
8
127.0.0.1:6379(TX)> set tx1 x1
QUEUED
127.0.0.1:6379(TX)> set tx2
(error) ERR wrong number of arguments for 'set' command
127.0.0.1:6379(TX)> set tx3 x3
QUEUED
127.0.0.1:6379(TX)> exec
(error) EXECABORT Transaction discarded because of previous errors.

组队阶段出现语法错误时，EXEC 会失败，事务中的所有命令都不会执行。

一部分成功、一部分失败

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> set tx1 x1
QUEUED
127.0.0.1:6379(TX)> set tx2 x2
QUEUED
127.0.0.1:6379(TX)> INCRBY tx2 1
QUEUED
127.0.0.1:6379(TX)> set tx3 x3
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
2) OK
3) (error) ERR value is not an integer or out of range
4) OK
127.0.0.1:6379>

如果命令成功入队，但执行时出现类型等运行错误，其他命令仍会继续执行。因此 Redis 事务不保证原子性，也不会自动回滚。

Redis 事务冲突解决方案

经典的抢票场景中，如果库存不足但多个请求同时扣减，没有并发控制就会造成超卖，库存甚至可能变为负数。

Redis 通过 WATCH 实现 check-and-set 风格的乐观锁：

乐观锁演示

Client A：

1
2
3
4
5
6
7
8
9
10
11
127.0.0.1:6379> set lock 100
OK
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set share_data ggstar
QUEUED
127.0.0.1:6379(TX)> incr lock
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
2) (integer) 101

Client B：

1
2
3
4
5
6
7
8
9
10
11
12
127.0.0.1:6379> get lock
"100"
127.0.0.1:6379> watch lock
OK
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set share_data mmstar
QUEUED
127.0.0.1:6379(TX)> incr lock
QUEUED
127.0.0.1:6379(TX)> exec
(nil)

Client A 先提交事务并修改了 lock。Client B 监视到版本已变化，因此 EXEC 返回 nil，事务执行失败。

火车票抢票案例（Java）

迭代 1：实现基本功能

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
public boolean doSecKill(String userId, String ticketNo) {
        // 库存key
        String stockKey = "sk:" + ticketNo + ":ticket";
        // 用户key, 购买的userId放入到这个set集合中
        String userKey = "sk:" + ticketNo + ":user";
        // 获取库存, 如果为null, 说明未配置库存, 直接返回
        Jedis jedis = getRedis();
        String stock = jedis.get(stockKey);
        if (StringUtils.isEmpty(stock)) {
            System.out.println("活动还未开始, 库存还未配置");
            release(jedis);
            return false;
        }
        // 判断用户是否重复购买
        boolean isUserBuy = jedis.sismember(userKey, userId);
        if (isUserBuy) {
            System.out.println(userId + " 已经参加过秒杀了");
            release(jedis);
            return false;
        }

        // 判断库存是否充足
        if (Integer.parseInt(stock) <= 0) {
            System.out.println("已售罄...");
            release(jedis);
            return false;
        }

        // 抢票扣减库存
        jedis.decr(stockKey);
        // 将已经抢到的用户加入set中, 用来判断是否重复购买
        jedis.sadd(userKey, userId);
        System.out.println(userId + "抢票成功");
        return true;
    }

问题分析：

• 判断库存和扣减库存不是原子操作。
• 多个线程可能同时判断库存充足，然后连续扣减，最终造成负库存。

迭代 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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
public boolean doSecKill(String userId, String ticketNo) {
        JedisPool jedisPool = JedisUtil.getInstance();
        Jedis jedis = jedisPool.getResource();

        // 库存key
        String stockKey = "sk:" + ticketNo + ":ticket";
        // 用户key, 购买的userId放入到这个set集合中
        String userKey = "sk:" + ticketNo + ":user";

        // 监视库存
        jedis.watch(stockKey);

        String stock = jedis.get(stockKey);
        if (stock == null) {
            System.out.println("秒杀还没有开始，请等待..");
            jedis.close();
            return false;
        }

        // 判断用户是否重复购买
        boolean isUserBuy = jedis.sismember(userKey, userId);
        if (isUserBuy) {
            System.out.println(userId + " 已经参加过秒杀了");
            jedis.close();
            return false;
        }

        // 判断库存是否充足
        if (Integer.parseInt(stock) <= 0) {
            System.out.println("已售罄...");
            jedis.close();
            return false;
        }

        // 开始事务
        Transaction multi = jedis.multi();
        multi.decr(stockKey);
        multi.sadd(userKey, userId);
        List<Object> result = multi.exec();
        if (CollectionUtils.isEmpty(result)) {
            System.out.println("抢票失败");
            jedis.close();
            return false;
        }
        return true;
    }

遗留问题：

• 600 张票、1000 个请求时，可能仍有库存没有卖出。
• 并发量越大，越多请求会因 WATCH 检测到版本变化而失败。

1
2
3
ab -n 1000 -c 100 -p ~/postfile \
  -T application/x-www-form-urlencoded \
  http://192.168.2.42:8080/seckill/seckill

迭代 3：使用 Lua 保证原子性

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
public boolean doSecKill(String userId, String ticketNo) {
JedisPool jedisPool = JedisUtil.getInstance();
Jedis jedis = jedisPool.getResource();

// 库存key
String stockKey = "sk:" + ticketNo + ":ticket";
// 用户key, 购买的userId放入到这个set集合中
String userKey = "sk:" + ticketNo + ":user";

String lua_script = null;
try {
    lua_script = IOUtils.resourceToString("seckill.lua", Charset.forName("utf8"), this.getClass().getClassLoader());
} catch (IOException e) {
    throw new RuntimeException(e);
}
// 将脚本加载到redis内存中
String sha1 = jedis.scriptLoad(lua_script);
// evalsha是根据sha1码来执行缓存在服务器的脚本
Object result = jedis.evalsha(sha1, 2, userId, ticketNo);
String resultStr = String.valueOf(result);
boolean ret = false;
switch (resultStr) {
    case "-1":
        System.out.println("活动未开始");
        ret = false;
    case "0":
        System.out.println("已售罄");
        ret = false;
    case "1":
        System.out.println("抢票成功");
        ret = true;
    case "2":
        System.out.println("用户已经购买过了");
        ret = false;
}
jedis.close();
return ret;
}

seckill.lua：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
local userid = KEYS[1];
local ticketno = KEYS[2];
local stockKey = 'sk:' .. ticketno .. ':ticket';
local usersKey = 'sk:' .. ticketno .. ':user';
if stockKey == nil then
    return -1
end
local userExists = redis.call('sismember', usersKey, userid);
if userExists ~= nil and tonumber(userExists) == 1 then
    return 2;
end
local num = redis.call("get", stockKey);
if tonumber(num) <= 0 then
    return 0;
else
    redis.call("decr", stockKey);
    redis.call("sadd", usersKey, userid);
end

return 1

Redis 连接池（Java）

定义

• 节省每次连接 redis 服务带来的消耗，把连接好的实例反复利用。

连接池参数

• MaxTotal：控制连接池最多可分配多少个 Jedis 实例；设置为 -1 表示不限制。
• maxIdle：控制连接池最多保留多少个空闲 Jedis 实例。
• MaxWaitMillis：获取 Jedis 实例时的最大等待毫秒数；超时后抛出 JedisConnectionException。
• testOnBorrow：获取 Jedis 实例时是否通过 ping() 检查连接可用性。

连接池工具类

Maven 依赖：

1
2
3
4
5
6
7
8
9
10
11
<!-- Spring 2.x 集成 Redis 所需 common-pool -->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
</dependency>

<!-- Jedis -->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

• 工具类,单例模式

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
public class JedisUtil {
    // volatile
    // 1.线程的可见性: 当一个线程去修改一个共线变量时, 另一外一个线程可以读取到整儿修改的值, 而不是副本
    // 2.禁止指令重排: 执行顺序和代码顺序一致
    private static volatile JedisPool jedisPool = null;

    private JedisUtil() {
    }

    // 单例模式-懒汉式-双检锁DCL
    public static JedisPool getInstance() {
        if (jedisPool == null) {
            synchronized (JedisUtil.class) {
                if (jedisPool == null) {
                    // 连接池配置
                    JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
                    // 最大连接数量
                    jedisPoolConfig.setMaxTotal(200);
                    // 最大空闲数量
                    jedisPoolConfig.setMaxIdle(32);
                    // 连接时最大等待时间60s
                    jedisPoolConfig.setMaxWaitMillis(60 * 1000);
                    // 耗尽连接时bocked
                    jedisPoolConfig.setBlockWhenExhausted(true);
                    // 检查连接是否可用
                    jedisPoolConfig.setTestOnBorrow(true);
                    jedisPool = new JedisPool(jedisPoolConfig, "192.168.2.85", 6379);
                }
            }
        }
        return jedisPool;
    }

    // 释放redis连接资源
    public static void release(Jedis jedis) {
        if (null != jedis) {
            // 如果jedis是从连接池中获取, 则close是把连接还给pool, 而不是销毁掉
            jedis.close();
        }
    }
}

• 使用

1
2
JedisPool jedisPool = JedisUtil.getInstance();
Jedis jedis = jedisPool.getResource();

Lua 脚本

定义

• 1、Lua 是一个小巧的脚本语言，Lua 脚本可以很容易的被 C/C++ 代码调用，也可以反过来调用 C/C++的函数，Lua 并没有提供强大的库，一个完整的 Lua 解释器不过 200k，所以 Lua 不适合作为开发独立应用程序的语言，而是作为嵌入式脚本语言。

• 2、很多应用程序、游戏使用 LUA 作为自己的嵌入式脚本语言，以此来实现可配置性、 可扩展性。

• 3、将复杂的或者多步的 Redis 操作，写为一个脚本，一次提交给 redis 执行，减少反复 连接 redis 的次数。提升性能。

• 4、LUA 脚本是类似 Redis 事务，有一定的原子性，不会被其他命令插队，可以完成一 些 redis 事务性的操作

• 5、Redis 的 lua 脚本功能，只有在 Redis 2.6 以上的版本才可以使用
• 6、通过 lua 脚本解决争抢问题，实际上是 Redis 利用其单线程的特性，用任务队列的 方式解决多任务并发问题

Lua 脚本的原子性

Lua 脚本类似 Redis 事务，具有一定的原子性。脚本执行期间不会被其他命令插队，可以完成事务性的操作。

使用 Lua 脚本解决争抢问题时，Redis 会利用单线程特性将请求形成任务队列，从而处理多任务并发问题。

Lua 脚本语法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
local userid=KEYS[1]; -- 获取传入的第一个参数
local ticketno=KEYS[2]; -- 获取传入的第二个参数
local stockKey='sk:'..ticketno..':ticket'; -- 拼接  stockKey
local usersKey='sk:'..ticketno..:user; -- 拼接 usersKey
local userExists=redis.call(sismember,usersKey,userid); -- 查看在 redis 的 usersKey set 中是否有该用户
if userExists ~= nil and tonumber(userExists)==1 then
    return 2; -- 如果该用户已经购买, 返回 2
end
local num= redis.call("get" ,stockKey); -- 获取剩余票数
if tonumber(num)<=0 then
   return 0; -- 如果已经没有票, 返回 0
else
    redis.call("decr",stockKey); -- 将剩余票数-1
    redis.call("sadd",usersKey,userid); -- 将抢到票的用户加入 set
end
return 1 -- 返回 1 表示抢票成功

• 参考：https://blog.csdn.net/qq_41286942/article/details/124161359
• lua脚本在java中写, 在redis中执行

Redis 调用 Lua 脚本

1
2
3
4
5
// 将脚本加载到redis内存中
String sha1 = jedis.scriptLoad(lua_script);
// evalsha是根据sha1码来执行缓存在服务器的脚本
Object result = jedis.evalsha(sha1, 2, userId, ticketNo);
String resultStr = String.valueOf(result);

Redis 主从复制

读写分离

工作方式

• 主节点数据更新后，会自动同步到从节点。
• Master 以写为主，Replica 以读为主，实现读写分离并降低单个 Redis 实例的压力。
• 从节点异常时可以切换到其他从节点，提高容灾恢复能力。
• 主从复制通常采用一主多从结构。
• 主节点的高可用可以通过 Sentinel 或 Redis Cluster 实现。

搭建一主多从

先复制基础 redis.conf 并关闭 AOF。主节点 6379 的配置如下：

1
2
3
4
include /zxredis/redis.conf
pidfile /var/run/redis_6379.pid
port 6379
dbfilename dump6379.rdb

• 从服务：6380、6381

1
2
3
4
include /zxredis/redis.conf
pidfile /var/run/redis_6380.pid
port 6380
dbfilename dump6380.rdb

• 在从服务上执行 SLAVEOF 127.0.0.1 6379，指定主服务。

• 使用 INFO replication 查看当前节点角色。

• 启动

1
2
3
redis-server redis_6379.conf
redis-server redis_6380.conf
redis-server redis_6381.conf

• 只能在主服务上进行写入操作, 在从服务写会报错, 写: 增加,删除,修改

主从复制原理

• Replica 成功连接 Master 后会发送同步命令。
• Master 启动后台持久化进程，同时收集新的写命令；持久化完成后，将完整数据文件发送给 Replica。
• Replica 接收数据文件后写入磁盘并加载到内存，这一过程属于全量复制。
• 后续 Master 会将新的写命令发送给 Replica，完成增量复制。
• Replica 重新连接 Master 时，可能再次触发全量复制。

一主二从

• 如果从服务器 down 了, 重新启动(需要重新执行SLAVEOF 127.0.0.1 6379) , 仍然可以获取 Master 的最新数据
• 如果主服务器 down 了, 从服务器并不会抢占为主服务器, 当主服务器恢复后, 从服务器仍然指向原来的主服务器.
• 主节点：6379
• 从节点：6380，执行 SLAVEOF 127.0.0.1 6379
• 从节点：6381，执行 SLAVEOF 127.0.0.1 6379

薪火相传

上一个 Slave 可以作为下一个 Slave 的 Master，并接收其他 Slave 的连接和同步请求。这种链式结构可以减轻主节点的写压力。使用 SLAVEOF <master_ip> <master_port> 指定上游节点。

这种结构的风险是：某个 Slave 宕机后，其下游节点将无法继续同步；主节点宕机后，从节点仍然是从节点，无法写入数据。

反客为主

• 在薪火相传的模式下,当一个 master 宕机后, 指向 Master 的 slave 可以升为 master, 其后面的 slave 不用做任何修改
• 用 slaveof no one 将从机变为主机 (后面可以使用哨兵模式, 自动完成切换.)

哨兵模式（Sentinel）

Sentinel 可以监控主节点状态。当主节点故障并满足判定条件时，Sentinel 会自动选择一个从节点升级为新的主节点。

配置步骤

1. 搭建一主二从结构：主节点为 6379，从节点为 6380、6381。
2. 创建 /zxredis/sentinel.conf：

1
sentinel monitor redis_master 127.0.0.1 6379 1

其中：

• redis_master 是被监控主节点的名称。
• 1 表示至少需要一个 Sentinel 同意故障转移。

1. 使用 redis-sentinel sentinel.conf 启动 Sentinel，默认端口为 26379。
2. 主节点故障后，Sentinel 从可用副本中选出新的主节点。
3. 原主节点恢复后，会作为新主节点的从节点重新加入。

新主节点选择顺序

Redis 集群

为什么需要集群：高可用与水平扩容

生产环境主要面临三个问题：

1. 单机容量不足时需要水平扩容。
2. 单个节点无法长期承担持续增长的读写压力。
3. 主从切换后，应用不应依赖固定的单机 IP 和端口。

Redis Cluster 将 key 映射到不同 slot，再把 slot 分配到不同主节点，从而分散容量和访问压力。生产环境中的主从节点应尽量部署在不同机器或故障域中。

集群搭建与使用

Redis Cluster 通过启动多个节点实现水平扩容，将数据分布到不同节点中。即使部分节点失效或无法通信，满足条件时集群仍可以继续处理请求。

搭建集群

1. 删除旧的 *.rdb 和 *.aof 文件。
2. 为每个节点创建独立配置文件。
3. 启动六个 Redis 实例。
4. 使用 redis-cli --cluster create 组建集群。

redis_6379.conf 示例：

1
2
3
4
5
6
7
include /zxredis/redis.conf
pidfile "/var/run/redis_6379.pid"
port 6379
dbfilename "dump6379.rdb"
cluster-enabled yes
cluster-config-file nodes-6379.conf
cluster-node-timeout 15000

复制配置文件：

1
2
3
4
5
cp redis_6379.conf redis_6380.conf
cp redis_6379.conf redis_6381.conf
cp redis_6379.conf redis_6389.conf
cp redis_6379.conf redis_6390.conf
cp redis_6379.conf redis_6391.conf

分别替换配置文件中的端口，例如在 Vim 中执行 :%s/6379/6381/g。

启动六个 Redis 服务：

1
2
3
4
5
6
redis-server redis_6379.conf
redis-server redis_6380.conf
redis-server redis_6381.conf
redis-server redis_6389.conf
redis-server redis_6390.conf
redis-server redis_6391.conf

将六个节点组成三主三从集群：

1
redis-cli --cluster create --cluster-replicas 1 192.168.2.85:6379 192.168.2.85:6380 192.168.2.85:6381 192.168.2.85:6389 192.168.2.85:6390 192.168.2.85:6391

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
>>> Performing hash slots allocation on 6 nodes...
Master[0] -> Slots 0 - 5460
Master[1] -> Slots 5461 - 10922
Master[2] -> Slots 10923 - 16383
[WARNING] Some slaves are in the same host as their master
M: ff1dd3947e4a180775f09b7257ca84518fae27ef 192.168.2.85:6379
   slots:[0-5460] (5461 slots) master
M: 841e721bc0f6c2378303b7544ae49e4fb70c7511 192.168.2.85:6380
   slots:[5461-10922] (5462 slots) master
M: 9ade621ffcf8bf2e9fee7f511892ef35b2a560b3 192.168.2.85:6381
   slots:[10923-16383] (5461 slots) master
S: b70b04658b23780962d1b9c1965f5a83ce7f3f61 192.168.2.85:6389
   replicates ff1dd3947e4a180775f09b7257ca84518fae27ef
S: 4af8384eff0836cbfd1d0bedf5f55d7961c709a9 192.168.2.85:6390
   replicates 841e721bc0f6c2378303b7544ae49e4fb70c7511
S: 070aa53b49980775f49b01c08a569bd7985f190d 192.168.2.85:6391
   replicates 9ade621ffcf8bf2e9fee7f511892ef35b2a560b3

注意：

• 组建集群前，确认所有实例已启动，并正常生成 nodes-xxxx.conf。
• 使用真实 IP，不要使用 127.0.0.1。
• --cluster-replicas 1 表示每个主节点配置一个从节点。
• 一个集群至少需要三个主节点。
• 主从节点应尽量部署在不同机器上，避免单机故障导致同一组主从同时失效。

使用 redis-cli -c -p 6379 连接集群，执行 CLUSTER NODES 查看节点和主从关系。

插槽

• 集群启动时会给 Master 节点分配插槽（slot）。
• Redis Cluster 包含 16384 个插槽，编号为 0-16383。
• Redis 使用 CRC16(key) % 16384 计算 key 所属的插槽。
• 每个 Master 节点负责一部分插槽。

在集群中写入数据

Redis 每次读写 key 时都会计算对应插槽。如果当前节点不负责该插槽，服务端会返回目标节点的地址和端口。

1
2
3
4
5
6
7
[root@localhost zxredis]# redis-cli -c -p 6379
127.0.0.1:6379> set k1 v1
-> Redirected to slot [12706] located at 192.168.2.85:6381
OK
192.168.2.85:6381> get k1
"v1"
192.168.2.85:6381>

redis-cli -c -p 6379 中的 -c 参数用于自动重定向。

不同 slot 下的 key 不能直接使用 MGET、MSET 等多 key 操作：

1
2
192.168.2.85:6381> mset k2 v2 k3 v3 k4 v4
(error) CROSSSLOT Keys in request don't hash to the same slot

可以使用 Hash Tag {} 让多个 key 分配到同一个 slot：

1
2
3
4
5
6
7
192.168.2.85:6381> mset k2{order} v2 k3{order} v3 k4{order} v4
OK
192.168.2.85:6381> keys *
1) "k1"
2) "k4{order}"
3) "k3{order}"
4) "k2{order}"

查询集群

CLUSTER KEYSLOT <key> 返回 key 对应的 slot：

1
2
192.168.2.85:6381> CLUSTER KEYSLOT k2{order}
(integer) 16025

CLUSTER COUNTKEYSINSLOT <slot> 返回 slot 中的 key 数量：

1
2
192.168.2.85:6381> CLUSTER COUNTKEYSINSLOT 16025
(integer) 3

CLUSTER GETKEYSINSLOT <slot> <count> 返回指定数量的 key：

1
2
3
4
192.168.2.85:6381> CLUSTER GETKEYSINSLOT 16025 3
1) "k2{order}"
2) "k3{order}"
3) "k4{order}"

集群故障恢复

• 主节点下线后，从节点会自动晋升为主节点。需要等待 cluster-node-timeout 判定超时。
• 原主节点恢复后，会作为从节点重新加入集群。 某一段插槽的主从节点全部不可用时，集群行为由 cluster-require-full-coverage 决定：

Jedis 集群开发

• 客户端即使连接的不是目标主节点，也可以根据重定向完成读写。
• 使用集群时，需要开放所有相关节点端口，否则可能出现 No more cluster attempts left。

示例代码：

1
2
3
4
5
6
HashSet<HostAndPort> hostAndPorts = new HashSet<>();
hostAndPorts.add(new HostAndPort("192.168.2.85", 6389));
JedisCluster jedisCluster = new JedisCluster(hostAndPorts);
jedisCluster.set("name", "hyy");
String name = jedisCluster.get("name");
System.out.println("name = " + name);

优点

• 支持水平扩容。
• 可以分摊访问压力。
• 无中心化配置相对简单。

缺点

• 不支持跨 slot 的多 key 操作。
• 不支持跨 slot 的 Redis 事务和 Lua 脚本。
• 从其他集群方案迁移到 Redis Cluster 时，通常需要整体迁移，复杂度较高。

缓存模型

请求链路：请求 -> Redis 缓存 -> 数据库。

缓存穿透

问题描述

• 前端请求查询了一个数据库中不存在的数据。连数据库里面都没有，那么缓存中肯定更没有
• 由于缓存穿透必然执行数据库查询，所以会增加数据库压力。严重时会把数据库压垮。

解决方案

缓存击穿

问题描述

数据库中存在对应数据，但 Redis 中的 key 已过期。请求无法命中缓存，只能查询数据库。

热点数据失效后，针对该数据的数据库访问会激增，显著增加数据库压力。

此时 Redis 表面运行正常，但数据库可能因瞬时压力过大而不可用。

解决方案

缓存雪崩

问题描述

大量缓存数据同时失效，原本可以从缓存获取数据的请求集中查询数据库，导致数据库压力激增。

严重时数据库会瞬间崩溃。

解决办法

分布式锁

问题描述

系统从单机部署演进为分布式集群后，请求会跨线程、进程和机器运行，单机 Java 锁无法协调所有节点。分布式锁通过跨 JVM 的互斥机制控制共享资源访问，只有成功获得锁的节点才能执行对应业务。

主流实现方案

基于 Redis 实现分布式锁

Java 实现

业务流程：

1. 尝试获取锁。
2. 获取成功后读取 num，执行加一并写回。
3. 在 finally 中校验锁值并释放锁。
4. 获取失败时等待 100 ms 后重试。

Java 代码

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
@RestController
public class RedisDistributedLock {
    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("/lock")
    public String lock() throws InterruptedException {
        // 获取随机的uuid避免锁被误删除
        String uuid = UUID.randomUUID().toString();
        // 设置3秒的过期时间,3秒后自动释放锁
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3, TimeUnit.SECONDS);
        if (lock) { // true说明获取到锁
            Object value = redisTemplate.opsForValue().get("num");
            // 没有值直接返回
            if (!StringUtils.hasText(value.toString())) {
                return "failed";
            }
            // 执行更新num+1的业务操作
            int num = Integer.parseInt(value.toString());
            redisTemplate.opsForValue().set("num", ++num);
            // 释放-lock,只有获取的锁和自己锁是同一把锁才删除
            String atomicDelLua = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
            redisScript.setScriptText(atomicDelLua);
            redisScript.setResultType(Long.class);
            Long result = (Long) redisTemplate.execute(redisScript, Arrays.asList("lock"), uuid);
            if (result.equals(1L)) {
                System.out.println("删除成功");
            }
            // -这里删除操作缺乏原子性,使用lua脚本原子性
            // if (uuid.equals(redisTemplate.opsForValue().get("lock"))) {
            //     redisTemplate.delete("lock");
            // }
            return "ok";
        } else {
            // 没有获取到锁, 睡眠100ms
            Thread.sleep(100);
            // 递归调用自己
            lock();
        }
        return "ok";
    }
}

Lua 解锁脚本

1
2
3
4
5
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
else
    return 0
end

压力测试

1
ab -n 1000 -c 100 http://192.168.2.42:8080/seckill/lock

分布式锁要求

锁的 key 应按业务资源划分。例如操作商品库存时，可以为每个 SKU 使用独立的锁。

Redis ACL（Redis 6）

Access Control List

ACL 可以限制用户能够执行的命令和能够访问的 key。

细粒度权限控制

ACL LIST

ACL LIST 展示所有用户及其权限规则。默认用户示例：

1
user default on nopass ~* &* +@all

ACL CAT

ACL CAT 查看命令类别，ACL CAT string 查看 String 类别包含的命令。

ACL WHOAMI

ACL WHOAMI 查看当前连接使用的用户名。

用户管理

创建 jack 后，权限规则大致如下：

1
user jack on #<password-hash> ~cached:* &* -@all +get

认证后尝试执行未授权命令会收到错误：

1
2
127.0.0.1:6379> set cached:k1 v1
(error) NOPERM this user has no permissions to run the 'set' command or its subcommand

Redis 6 新特性

I/O 多线程

I/O 多线程是指与客户端交互时，网络 I/O 处理模块使用多线程，并不是使用多线程执行 Redis 命令。

Redis 6 执行命令时仍然是单线程。

客户端交互采用多线程，命令执行仍然是单线程配合 I/O 多路复用。

I/O 多线程默认不开启，需要在配置文件中设置：

1
2
io-threads-do-reads no
io-threads 4

Cluster 工具

• 旧版本 Redis 搭建集群需要 Ruby 环境。
• Redis 5 将 redis-trib.rb 的功能集成到了 redis-cli。
• redis-benchmark 是性能测试工具，可以用于集群压测。

RESP3

RESP3 是新的 Redis 通信协议，用于优化服务端和客户端之间的通信。

客户端缓存

客户端缓存基于 RESP3 协议实现，可以提高性能并减少 TCP 网络交互。

集群代理

代理模式让 Cluster 拥有类似单实例的接入方式，降低使用集群的门槛。

Modules API

• 上一篇
  前端基础学习笔记
• 下一篇
  精神现象学