day09 微服务阶段面试篇下 ​

Java工程师高频面题：https://yjoffer.com/javaqa/readme.html

学习目标 ​

1. 能够说出xxl-job任务调度的优势
2. 能说出xxl-job的组成结构
3. 能够编写热点商品更新缓存任务
4. 能够说出Redid的持久化机制
5. 能够说出Redis主从同步原理
6. 能够说出Redis哨兵模式的工作原理
7. 能够说出Redis分片集群的工作原理
8. 能够说出Redis五种基本数据类型
9. 能够说出Redis ZSET类型的底层结构
10. 能够说出Redis过期策略
11. 能够说出Redis淘汰策略

🎬 视频播放链接

1 任务调度方案 ​

1.1 什么是任务调度 ​

1.1.1 概念 ​

我们可以先思考一下下面业务场景的解决方案：

某电商系统需要在每天上午10点，下午3点，晚上8点发放一批优惠券。

某银行系统需要在信用卡到期还款日的前三天进行短信提醒。

某财务系统需要在每天凌晨0:10结算前一天的财务数据，统计汇总。

12306会根据车次的不同，而设置某几个时间点进行分批放票。

某网站为了实现天气实时展示，每隔5分钟就去天气服务器获取最新的实时天气信息。

以上场景就是任务调度所需要解决的问题。

任务调度是指系统为了自动完成特定任务，在约定的特定时刻去执行任务的过程。有了任务调度即可解放更多的人 力由系统自动去执行任务。

在解决缓存击穿方案中，通过缓存定时预热避免缓存击穿。

1.1.2 技术方案 ​

如何实现任务调度？

1、使用jdk提供的Timer定时器

示例代码如下：

每个Timer对应一个线程，可以同时启动多个Timer定时执行多个任务。

public static void main(String[] args){  
    Timer timer = new Timer();  
    timer.schedule(new TimerTask(){
@Override  
public void run() {  
   //TODO：something
}  
    }, 1000, 2000);  //1秒后开始调度，每2秒执行一次
}

Time使用简单，可以实现每隔一定的时间去执行任务，但无法实现每天凌晨去执行任务，即在某个时间点去执行任务。

2、使用第三方Quartz方式实现

Quartz 是一个功能强大的任务调度框架（项目地址：https://github.com/quartz-scheduler/quartz ），它可以满足更多更复杂的调度需求，Quartz 设计的核心类包括 Scheduler, Job 以及 Trigger。其中，Job 负责定义需要执行的任务，Trigger 负责设置调度策略，Scheduler 将二者组装在一起，并触发任务开始执行。Quartz支持简单的按时间间隔调度、还支持按日历调度方式，通过设置CronTrigger表达式（包括：秒、分、时、日、月、周、年）进行任务调度。

虽然Quartz可以实现按日历调度的方式，但无法支持分布式环境下任务调度。分布式环境下通常一个服务部署多个实例即多个jvm进程，假设一个项目的微服务部署两个实例每个实例定时执行更新缓存的任务，两个实例就会重复执行。如下图：

3、使用分布式调度平台XXL-JOB

XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。

官网：https://www.xuxueli.com/xxl-job/

文档：https://www.xuxueli.com/xxl-job/#《分布式任务调度平台XXL-JOB》

XXL-JOB主要有调度中心、执行器、任务：

调度中心：

负责管理调度信息，按照调度配置发出调度请求，自身不承担业务代码；

主要职责为执行器管理、任务管理、监控运维、日志管理等

任务执行器：

负责接收调度请求并执行任务逻辑；

主要职责是执行任务、执行结果上报、日志服务等

使用XXL-JOB可以解决多个jvm进程重复执行任务的问题，如下图：

XXL-JOB调度中心可以配置路由策略，比如：第一个、轮询策略、分片等，它们分别表示的意义如下：

第一个：即每次执行任务都由第一个执行器去执行。

轮询：即执行器轮番执行。

分片：每次执行任务广播给每个执行器让他们同时执行任务。

如果根据需求每次执行任务仅由一个执行器去执行任务可以设置路由策略：第一个、轮询。

如果根据需求每次执行任务由多个执行器同时执行可以设置路由策略为：分片。

xxl-job分布式任务调度系统具体有以下优势：

1、并行任务调度

并行任务调度实现靠多线程，如果有大量任务需要调度，此时光靠多线程就会有瓶颈了，因为一台计算机CPU的处理能力是有限的。

如果将任务调度程序分布式部署，每个结点还可以部署为集群，这样就可以让多台计算机共同去完成任务调度，我们可以将任务分割为若干个分片，由不同的实例并行执行，来提高任务调度的处理效率。

2、高可用

若某一个实例宕机，不影响其他实例来执行任务。

3、弹性扩容

当集群中增加实例就可以提高并执行任务的处理效率。

4、任务管理与监测

对系统中存在的所有定时任务进行统一的管理及监测。让开发人员及运维人员能够时刻了解任务执行情况，从而做出快速的应急处理响应。

5、避免任务重复执行

当任务调度以集群方式部署，同一个任务调度可能会执行多次，比如在上面提到的电商系统中到点发优惠券的例子，就会发放多次优惠券，对公司造成很多损失，所以我们需要控制相同的任务在多个运行实例上只执行一次。

1.1.3 小结 ​

xxl-job任务调度与第三方Quartz或timer定时器实现任务调度有什么优势？

1.2 搭建XXL-JOB ​

1.2.1 组成结构 ​

XXL-JOB由两部分组成：

• 调度模块（调度中心）：
  负责管理调度信息，按照调度配置发出调度请求，自身不承担业务代码。调度系统与任务解耦，提高了系统可用性和稳定性，同时调度系统性能不再受限于任务模块；
  支持可视化、简单且动态的管理调度信息，包括任务新建，更新，删除，GLUE开发和任务报警等，所有上述操作都会实时生效，同时支持监控调度结果以及执行日志，支持执行器Failover。
• 执行模块（执行器）：
  负责接收调度请求并执行任务逻辑。任务模块专注于任务的执行等操作，开发和维护更加简单和高效；
  接收“调度中心”的执行请求、终止请求和日志请求等。

1.2.2 部署调度中心 ​

1.查阅xxl-job的源码

首先下载XXL-JOB

GitHub：https://github.com/xuxueli/xxl-job

码云：https://gitee.com/xuxueli0323/xxl-job

项目使用2.3.1版本： https://github.com/xuxueli/xxl-job/releases/tag/2.3.1

也可从课程资料目录获取，解压xxl-job-2.3.1.zip

使用IDEA打开解压后的目录

xxl-job-admin：调度中心

xxl-job-core：公共依赖

xxl-job-executor-samples：执行器Sample示例（选择合适的版本执行器，可直接使用）

：xxl-job-executor-sample-springboot：Springboot版本，通过Springboot管理执行器，推荐这种方式；

：xxl-job-executor-sample-frameless：无框架版本；

doc :文档资料，包含数据库脚本

在下发的虚拟机的MySQL中已经创建了xxl_job_2.3.1数据库

如下图：

安装xxl-job ​

没有使用下发虚拟机的同学请自行安装xxl-job。

拉取镜像：

docker pull xuxueli/xxl-job-admin:2.3.1

创建数据库：xxl_job_2.3.1

导入xxl_job_2.3.1.sql,如下：

创建目录：

/data/soft/xxl-job

/data/soft/xxl-job/applogs

创建配置文件：/data/soft/xxl-job/application.properties,内容如下：

### web
server.port=8080
server.servlet.context-path=/xxl-job-admin

### actuator
management.server.servlet.context-path=/actuator
management.health.mail.enabled=false

### resources
spring.mvc.servlet.load-on-startup=0
spring.mvc.static-path-pattern=/static/**
spring.resources.static-locations=classpath:/static/

### freemarker
spring.freemarker.templateLoaderPath=classpath:/templates/
spring.freemarker.suffix=.ftl
spring.freemarker.charset=UTF-8
spring.freemarker.request-context-attribute=request
spring.freemarker.settings.number_format=0.##########

### mybatis
mybatis.mapper-locations=classpath:/mybatis-mapper/*Mapper.xml
#mybatis.type-aliases-package=com.xxl.job.admin.core.model

### xxl-job, datasource
spring.datasource.url=jdbc:mysql://192.168.101.68:3306/xxl_job_2.3.1?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&serverTimezone=Asia/Shanghai
spring.datasource.username=root
spring.datasource.password=mysql
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

### datasource-pool
spring.datasource.type=com.zaxxer.hikari.HikariDataSource
spring.datasource.hikari.minimum-idle=10
spring.datasource.hikari.maximum-pool-size=30
spring.datasource.hikari.auto-commit=true
spring.datasource.hikari.idle-timeout=30000
spring.datasource.hikari.pool-name=HikariCP
spring.datasource.hikari.max-lifetime=900000
spring.datasource.hikari.connection-timeout=10000
spring.datasource.hikari.connection-test-query=SELECT 1
spring.datasource.hikari.validation-timeout=1000

### xxl-job, email
spring.mail.host=smtp.qq.com
spring.mail.port=25
spring.mail.username=xxx@qq.com
spring.mail.from=xxx@qq.com
spring.mail.password=xxx
spring.mail.properties.mail.smtp.auth=true
spring.mail.properties.mail.smtp.starttls.enable=true
spring.mail.properties.mail.smtp.starttls.required=true
spring.mail.properties.mail.smtp.socketFactory.class=javax.net.ssl.SSLSocketFactory

### xxl-job, access token
xxl.job.accessToken=default_token

### xxl-job, i18n (default is zh_CN, and you can choose "zh_CN", "zh_TC" and "en")
xxl.job.i18n=zh_CN

## xxl-job, triggerpool max size
xxl.job.triggerpool.fast.max=200
xxl.job.triggerpool.slow.max=100

### xxl-job, log retention days
xxl.job.logretentiondays=30

创建容器：

docker run -d -e \
  --restart=always \
  -v /data/soft/xxl-job/applogs:/data/applogs \
  -v /data/soft/xxl-job/application.properties:/application.properties \
  -p 8088:8080 \
  --name xxl-job-admin \
  xuxueli/xxl-job-admin:2.3.1

启动成功进入管理界面：

http://192.168.101.68:8088/xxl-job-admin

账号/密码：admin/123456

2.启动xxl-job

执行docker start xxl-job-admin 启动xxl-job

访问：http://192.168.101.68:8088/xxl-job-admin/

账号和密码：admin/123456

1.2.3 执行器 ​

1.添加执行器依赖

下边配置执行器，执行器负责与调度中心通信接收调度中心发起的任务调度请求，执行器负责执行微服务中定义的任务，执行器程序由xxl-job提供，在微服务中引入下边的依赖即加入了执行器的程序。

我们在商品服务中引入xxl-job执行器依赖。

<dependency>
    <groupId>com.xuxueli</groupId>
    <artifactId>xxl-job-core</artifactId>
    <version>2.3.0</version>
</dependency>

参考源代码中的XxlJobConfig去编写xxl-job的配置类，此配置类已提供，将课程资料中xxl-job下的配置类和模型类拷贝到商品服务的config包下：

2.配置xxl-job

在application.yaml下配置xxl-job

xxl-job:
  enable: true
  port: 11603
  access-token: default_token
  admin:
    address: http://192.168.101.68:8088/xxl-job-admin
  executor:
    appName: ${spring.application.name}
    #ip: 172.17.0.170
    port: ${xxl-job.port}
    # 执行器日志文件保存天数 [选填] ： 过期日志自动清理, 限制值大于等于3时生效; 否则, 如-1, 关闭自动清理功能
    log-retention-days: 30

说明：

address：调度中心的地址

appName：执行器名称，spring.application.name表示微服务的名称（在bootstrap.yml中配置）

port：执行器端口号,通过xxl-job.port配置，执行器通过此端口与调度中心通信。

3. 下边进入调度中心添加执行器

启动商品服务即启动了xxl-job执行器。

进入调度中心，进入执行器管理界面，如下图：

点击新增，填写执行器信息

AppName：执行名称， appName: ${spring.application.name}表示指定执行器名称就是微服务的应用名。

名称：取一个中文名称。

注册方式：自动注册，只要执行器和调度中心连通执行器会自动注册到调度中心

机器地址：自动注册时不用填写。

添加成功：

启动item-service，查看item-service的控制台：

>>>>>>>>>>> xxl-job remoting server start success, nettype = class com.xxl.job.core.server.EmbedServer, port = 11603 说明执行器启动成功。

稍等片刻进入 xxl-job调度中心，进入执行器管理界面，执行器注册成功：

点击“查看（1）”，查看执行器的地址，如下图：

1.2.4 小结 ​

项目为什么要用xxl-job？

能说出xxl-job的组成部分。

1.3 XXL-JOB任务入门 ​

1.3.1 编写测试任务 ​

定时执行任务就需要编写任务方法，此任务方法由执行器去调用。

可以参考xxl-job源码去编写任务方法，从源码目录中找到执行器示例代码：

xxl-job-2.3.1\xxl-job-executor-samples\xxl-job-executor-sample-springboot\src\main\java\com\xxl\job\executor\service\jobhandler\SampleXxlJob.java

部分示例代码如下：下边代码中demoJobHandler()就是一个任务方法，需要使用@XxlJob注解标识，所在类需要由spring去管理，所以加了@Component注解。

将源代码中的SampleXxlJob类拷贝到商品服务的job包下，修改代码如下：

package com.hmall.item.job;
...
@Component
@Slf4j
public class SampleXxlJob {
    private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);


    /**
     * 1、简单任务示例（Bean模式）
     */
    @XxlJob("demoJobHandler")
    public void demoJobHandler() throws Exception {
log.info("XXL-JOB, Hello World.");

for (int i = 0; i < 5; i++) {
    log.info("beat at:" + i);
    TimeUnit.SECONDS.sleep(2);
}
// default success
    }


    /**
     * 2、分片广播任务
     */
    @XxlJob("shardingJobHandler")
    public void shardingJobHandler() throws Exception {

// 分片参数
int shardIndex = XxlJobHelper.getShardIndex();
int shardTotal = XxlJobHelper.getShardTotal();

log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);

// 业务逻辑
for (int i = 0; i < shardTotal; i++) {
    if (i == shardIndex) {
log.info("第 {} 片, 命中分片开始处理", i);
    } else {
log.info("第 {} 片, 忽略", i);
    }
}

    }

}

1.3.2 配置任务 ​

下边在调度中心配置任务。

进入任务管理，新增任务：

填写任务信息：

说明：

调度类型：

固定速度指按固定的间隔定时调度。

Cron，通过Cron表达式实现更丰富的定时调度策略。

Cron表达式是一个字符串，通过它可以定义调度策略，格式如下：

{秒数} {分钟} {小时} {日期} {月份} {星期}

Cron 的各个域的定义如下表格所示：

xxl-job提供图形界面去配置： 或者使用yjoffer.com在线工具生成Cron表达式，非常方便

一些例子如下：

0 0 0 * * ? 每天0点触发

30 10 1 * * ? 每天1点10分30秒触发

0/30 * * * * ? 每30秒触发一次

* 0/10 * * * ? 每10分钟触发一次

为了方便测试这里第5秒执行一次，设置为：0/5 * * * * ?

cron 表达式的难点在于通配符，下边的内容请自行阅读

• , 这里指的是在两个以上的时间点中都执行，如果我们在 “分” 这个域中定义为 8,12,35 ，则表示分别在第8分，第12分 第35分执行该定时任务。
• - 这个比较好理解就是指定在某个域的连续范围，如果我们在 “时” 这个域中定义 1-6，则表示在1到6点之间每小时都触发一次，用 , 表示 1,2,3,4,5,6
• * 表示所有值，可解读为 “每”。 如果在“日”这个域中设置 *,表示每一天都会触发。
• ? 表示不指定值。使用的场景为不需要关心当前设置这个字段的值。例如:要在每月的8号触发一个操作，但不关心是周几，我们可以这么设置 0 0 0 8 * ?
• / 在某个域上周期性触发，该符号将其所在域中的表达式分为两个部分，其中第一部分是起始值，除了秒以外都会降低一个单位，比如 在 “秒” 上定义 5/10 表示从 第 5 秒开始 每 10 秒执行一次，而在 “分” 上则表示从 第 5 秒开始 每 10 分钟执行一次。
• L 表示英文中的LAST 的意思，只能在 “日”和“周”中使用。在“日”中设置，表示当月的最后一天(依据当前月份，如果是二月还会依据是否是润年), 在“周”上表示周六，相当于”7”或”SAT”。如果在”L”前加上数字，则表示该数据的最后一个。例如在“周”上设置”7L”这样的格式,则表示“本月最后一个周六”
• W 表示离指定日期的最近那个工作日(周一至周五)触发，只能在 “日” 中使用且只能用在具体的数字之后。若在“日”上置”15W”，表示离每月15号最近的那个工作日触发。假如15号正好是周六，则找最近的周五(14号)触发, 如果15号是周未，则找最近的下周一(16号)触发.如果15号正好在工作日(周一至周五)，则就在该天触发。如果是 “1W” 就只能往本月的下一个最近的工作日推不能跨月往上一个月推。
• # 表示每月的第几个周几，只能作用于 “周” 上。例如 ”2#3” 表示在每月的第三个周二。

运行模式有BEAN和GLUE，bean模式较常用就是在项目工程中编写执行器的任务代码，GLUE是将任务代码编写在调度中心。

JobHandler即任务方法名，填写任务方法上边@XxlJob注解中的名称。

路由策略：

第一个：即每次执行任务都由第一个执行器去执行。

轮询：即执行器轮番执行。

分片：每次执行任务广播给每个执行器让他们同时执行任务。

详细说明xxl-job源码中的doc目录下的文档：

1.3.3 启动任务并测试 ​

任务配置完成，下边启动任务

启动成功：

我们在任务方法上打断点跟踪，任务方法被执行，如下图：

1.4 分片广播任务 ​

1.4.1 什么是分片广播任务 ​

掌握了xxl-job的基本使用，下边思考如何进行分布式任务处理呢？如下图，我们会启动多个执行器组成一个集群，去执行任务。

查看xxl-job官方文档，阅读高级配置相关的内容：

高级配置：
    - 路由策略：当执行器集群部署时，提供丰富的路由策略，包括；
FIRST（第一个）：固定选择第一个机器；
LAST（最后一个）：固定选择最后一个机器；
ROUND（轮询）：；
RANDOM（随机）：随机选择在线的机器；
CONSISTENT_HASH（一致性HASH）：每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上。
LEAST_FREQUENTLY_USED（最不经常使用）：使用频率最低的机器优先被选举；
LEAST_RECENTLY_USED（最近最久未使用）：最久未使用的机器优先被选举；
FAILOVER（故障转移）：按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度；
BUSYOVER（忙碌转移）：按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度；
SHARDING_BROADCAST(分片广播)：广播触发对应集群中所有机器执行一次任务，同时系统自动传递分片参数；可根据分片参数开发分片任务；

下边要重点说的是分片广播策略，分片是指是调度中心以执行器为维度进行分片，将集群中的执行器标上序号：0，1，2，3...，广播是指每次调度会向集群中的所有执行器发送任务调度，请求中携带分片参数。

分片广播任务就是调度中心按照调度策略广播通信所有执行器（分片）去执行任务。

如下图：

每个执行器收到调度请求同时接收分片参数。

xxl-job支持动态扩容执行器集群从而动态增加分片数量，当有任务量增加可以部署更多的执行器到集群中，调度中心会动态修改分片的数量。

作业分片适用哪些场景呢？

• 分片任务场景：10个执行器的集群来处理10w条数据，每台机器只需要处理1w条数据，耗时降低10倍；

所以，广播分片方式不仅可以充分发挥每个执行器的能力，并且根据分片参数可以控制任务是否执行，最终灵活控制了执行器集群分布式处理任务。

使用说明：

"分片广播" 和普通任务开发流程一致，不同之处在于可以获取分片参数进行分片业务处理。

Java语言任务获取分片参数方式：

BEAN、GLUE模式(Java)，可参考Sample示例执行器中的示例任务"ShardingJobHandler"：

/**
 * 2、分片广播任务
 */
@XxlJob("shardingJobHandler")
public void shardingJobHandler() throws Exception {

    // 分片参数
    int shardIndex = XxlJobHelper.getShardIndex();
    int shardTotal = XxlJobHelper.getShardTotal();

    XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);

    // 业务逻辑
    for (int i = 0; i < shardTotal; i++) {
if (i == shardIndex) {
    XxlJobHelper.log("第 {} 片, 命中分片开始处理", i);
} else {
    XxlJobHelper.log("第 {} 片, 忽略", i);
}
    }

}

1.4.2 测试分片广播任务 ​

下边测试作业分片广播任务：

1、定义作业分片的任务方法

/**
 * 2、分片广播任务
 */
@XxlJob("shardingJobHandler")
public void shardingJobHandler() throws Exception {

    // 分片参数
    int shardIndex = XxlJobHelper.getShardIndex();
    int shardTotal = XxlJobHelper.getShardTotal();

    XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);

    // 业务逻辑
    for (int i = 0; i < shardTotal; i++) {
if (i == shardIndex) {
    XxlJobHelper.log("第 {} 片, 命中分片开始处理", i);
} else {
    XxlJobHelper.log("第 {} 片, 忽略", i);
}
    }

}

2、在调度中心添加任务

添加成功：

下边启动两个商品服务实例

两个实例的在启动时注意端口不能冲突：

实例1 在VM options处添加：-Dserver.port=8081 -Dxxl-job.port=11603

实例2 在VM options处添加：-Dserver.port=7081 -Dxxl-job.port=11604

启动成功观察执行器

启动任务，观察日志

实例1:

实例2：

下边启动两个执行器实例，观察每个实例的执行情况。

1.4.3. 分片执行任务 ​

当一次分片广播到来，各执行器如何根据分片参数去分布式执行任务，保证执行器之间执行的任务不重复呢？

举例：

批量处理商品表中的数据，保证每个执行器处理的商品信息不重复。

可以将分片总数和分片序列带入sql，如下：

select * from item where id % #{shardingTotalCount} = #

假设当前有两个分片，分片0执行如下sql:

select * from item where id % 2=0

分片1执行如下sql:

select * from item where id % 2=1

两个分片获取的数据是不一样的。

测试如下：

定义mapper如下：

public interface ItemMapper extends BaseMapper<Item> {
   ....

    //根据分片总数和分片序号查询商品表 select * from item where id % ? = ?
    @Select("select * from item where id % #{shardingTotalCount} = #{shardingIndex}")
    List<Item> selectBySharding(int shardingTotalCount, int shardingIndex);
}

在任务中调用ItemMapper获取商品信息

@XxlJob("shardingJobHandler")
public void shardingJobHandler() throws Exception {

    // 分片参数
    int shardIndex = XxlJobHelper.getShardIndex();
    int shardTotal = XxlJobHelper.getShardTotal();

    //查询商品表
    //select * from item where id % ? = ?
    List<Item> items = itemMapper.selectBySharding(shardTotal, shardIndex);

    log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
    }

1.5 热点商品定时预热任务 ​

1.5.1 编写任务方法 ​

根据需求，为了防止缓存击穿我们使用xxl-job定时对热点商品进行预热。

首先在ItemServiceImpl中编写获取热点商品id方法

@Override
public List<Long> queryHotItems() {
    //模拟热点商品id
    return List.of(317578L, 317580L);
}

编写商品预热任务方法

@Component
@Slf4j
public class SampleXxlJob {
    private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);

    //注入itemService
    @Resource
    private IItemService itemService;
    
/**
 * 对热点商品进行定时更新缓存
 */
@XxlJob("hotItemCacheJobHandler")
public void hotItemCacheJobHandler() throws Exception {
    log.info("定时对热点商品进行更新缓存开始...");
    //获取热点商品id
    List<Long> ids = itemService.queryHotItems();
    itemService.queryItemByIdsCache(ids);
    log.info("定时对热点商品进行更新缓存完成...");
}
...

1.5.2 配置任务 ​

下边在调度中心配置任务。

进入任务管理，新增任务：

1.5.3 启动任务并测试 ​

任务配置完成，下边启动任务

下边重启商品服务，观察热点商品是否在redis更新缓存。

1.5.4 小结 ​

项目中哪里用了xxl-job？怎么用的？

2 Redis 持久化 ​

2.1 面试题 ​

面试题：说一下Redid的持久化机制。

该面试题考察你对Redis持久化机制的理解程度。

Redis使用两种主要的持久化机制来保证数据的安全性：RDB（Redis Database Backup）和AOF（Append Only File），需要去理解这两种持久化机制的工作原理及应用场景才能回答本问题。

2.2 RDB ​

2.2.1 RDB是什么？ ​

RDB全称Redis Database Backup file（Redis数据备份文件），也被叫做Redis数据快照。简单来说就是把内存中的所有数据都记录到磁盘中。当Redis实例故障重启后，从磁盘读取快照文件，恢复数据。快照文件称为RDB文件。

2.2.2 执行方式 ​

RDB 的触发方式包括：手动触发、自动触发。

1. 手动触发：

   • 使用SAVE命令，这会导致Redis在执行完该命令之前阻塞客户端请求。
   • 使用BGSAVE命令，Redis会在后台异步进行快照操作，不会阻塞客户端请求。

2. 自动触发：

   • 通过配置save选项，可以设置在满足特定条件时自动触发RDB持久化。例如，可以在一定时间内发生了一定数量的写操作后自动进行一次快照。
   • Redis停机时，Redis停机时会执行一次save命令，实现RDB持久化。

在redis.conf文件中配置了rdb的文件名称和保存路径：

是否压缩 ,建议不开启，压缩也会消耗cpu，磁盘的话不值钱
rdbcompression yes
RDB文件名称
dbfilename dump.rdb  
文件保存的路径目录
dir ./

1）save命令

执行save命令可以立即执行一次RDB,在执行完该命令之前阻塞客户端请求.

进入redis容器：docker exec -it redis /bin/bash

在执行命令前我查看rdb文件

执行save命令后从文件更新时间上看发现rdb文件有更新

2）bgsave命令

下面的命令可以异步执行RDB：

这个命令执行后会开启独立进程完成RDB，主进程可以持续处理用户请求，不受影响。

3）触发RDB条件

Redis内部有触发RDB的机制，可以在redis.conf文件中找到，格式如下：

900秒内，如果至少有1个key被修改，则执行bgsave ， 如果是save "" 则表示禁用RDB
save 900 1  
save 300 10  
save 60 10000

2.2.3 执行原理 ​

RDB 的工作流程如下图所示：

1. 启动子进程：当Redis服务器接收到BGSAVE命令时，它会先检查是否有正在运行的子进程。如果没有，则创建一个新的子进程（fork操作）。
2. 复制父进程内存映像：此时父进程和子进程共享相同的内存页。当任一进程对某个内存页进行修改时，系统才会为这个进程分配新的内存页并复制内存过去。
3. 子进程生成RDB文件：子进程开始读取内存中的数据并生成RDB文件。在此过程中，父进程继续处理客户端请求。
4. 替换旧的RDB文件：一旦子进程完成RDB文件的生成，它会用新生成的文件替换旧的RDB文件。
5. 父进程继续工作：子进程结束后，父进程继续正常工作，此时新的RDB文件已经包含了最新的数据状态。

2.2.4 RDB 的优缺点 ​

• 优点：

  • 性能影响小：因为RDB文件的生成是在子进程中进行的，不会阻塞主进程。
  • 启动速度快：在Redis重启时，可以通过加载RDB文件快速恢复数据。
  • 文件紧凑：RDB文件是一个紧凑的二进制文件，占用的空间较小。

• 缺点：

  • 数据丢失风险：如果Redis服务器在最后一次成功生成RDB文件后宕机，那么从上次快照以来的所有更改都将丢失。
  • 快照生成频率：需要平衡性能影响与数据丢失的风险。

2.2.5 总结 ​

RDB方式bgsave的基本流程？

• fork主进程得到一个子进程，共享内存空间
• 子进程读取内存数据并写入新的RDB文件
• 用新RDB文件替换旧的RDB文件

RDB会在什么时候执行？save 60 1000代表什么含义？

• 默认是服务停止时
• 代表60秒内至少执行1000次修改则触发RDB

RDB的缺点？

• RDB执行间隔时间长，两次RDB之间写入数据有丢失的风险
• fork子进程、压缩、写入RDB文件都比较耗时

2.3 AOF ​

2.3.1.AOF原理 ​

AOF全称为Append Only File（追加文件）是Redis提供的另一种持久化机制，与RDB相比，AOF持久化通过记录每次写操作的命令到一个单独的日志文件中，使得即使Redis服务重启，也可以通过重放这些命令来恢复数据。

2.3.2.AOF配置 ​

AOF默认是关闭的，需要修改redis.conf配置文件来开启AOF：

#是否开启AOF功能，默认是no
appendonly yes
#AOF文件的名称
appendfilename "appendonly.aof"

AOF的命令记录的频率也可以通过redis.conf文件来配：

#表示每执行一次写命令，立即记录到AOF文件
appendfsync always 
#写命令执行完先放入AOF缓冲区，然后表示每隔1秒将缓冲区数据写到AOF文件，是默认方案
appendfsync everysec 
#写命令执行完先放入AOF缓冲区，由操作系统决定何时将缓冲区内容写回磁盘
appendfsync no

三种策略对比：

打开aof重启redis，写入数据查看aof文件的内容，aof内容是若干redis命令的集合。

2.3.3.AOF文件重写 ​

因为是记录命令，AOF文件会比RDB文件大的多。而且AOF会记录对同一个key的多次写操作，但只有最后一次写操作才有意义。通过执行bgrewriteaof命令，可以让AOF文件执行重写功能，用最少的命令达到相同效果。

如图，AOF原本有三个命令，但是set num 123 和 set num 666都是对num的操作，第二次会覆盖第一次的值，因此第一个命令记录下来没有意义。

所以重写命令后，AOF文件内容就是：mset name jack num 666

Redis也会在触发阈值时自动去重写AOF文件。阈值也可以在redis.conf中配置：

# AOF文件比上次文件 增长超过多少百分比则触发重写
auto-aof-rewrite-percentage 100
# AOF文件体积最小多大以上才触发重写 
auto-aof-rewrite-min-size 64mb

2.4 总结 ​

RDB与AOF对比，RDB和AOF各有自己的优缺点，如下图：

在实际应用中，RDB（Redis Database dump）和AOF（Append Only File）这两种持久化方式各有优势，通常的选择取决于具体的应用场景和需求。

如果对数据完整性有极高要求，那么AOF通常是首选；

如果对性能要求较高并且可以接受少量的数据丢失，那么RDB是更好的选择。

而在大多数情况下，结合使用RDB和AOF能够提供最佳的数据保护和性能表现。

RDB与AOF详细总结如下：

1. RDB (Redis Database dump)

• 原理

  • 是把内存中的所有数据都记录到磁盘中。当Redis实例故障重启后，从磁盘读取快照文件，恢复数据。快照文件称为RDB文件。
  • 执行方式：手动触发，自动触发
  • 异步方式同步，当Redis服务器接收到BGSAVE命令时，它会先检查是否有正在运行的子进程。如果没有，则创建一个新的子进程（fork操作）。子进程开始读取内存中的数据并生成RDB文件。在此过程中，父进程继续处理客户端请求。一旦子进程完成RDB文件的生成，它会用新生成的文件替换旧的RDB文件。

• 主要用途：

  • 数据备份：由于RDB文件是某个时间点的快照，因此非常适合用于数据备份和灾难恢复。
  • 快速恢复：RDB文件相对较小，可以在服务重启时快速加载到内存中，适合需要快速恢复服务的情况。

• 适用场景：

  • 当数据的完整性不是绝对重要，可以接受一定程度的数据丢失时。
  • 对于读多写少的应用，RDB可以提供较好的性能。
  • 需要定期做全量备份时。

• 配置示例：

  • 使用save命令手动触发快照生成。
  • 设置自动快照生成规则，例如：save 900 1 表示在900秒内如果有1个key发生变化，则生成快照。

1. AOF (Append Only File)

• 原理

  • AOF持久化通过记录每次写操作的命令到一个单独的日志文件中，使得即使Redis服务重启，也可以通过重放这些命令来恢复数据。

• 主要用途：

  • 数据持久化：AOF通过记录每个写命令来保证数据的完整性，非常适合需要强一致性的场景。
  • 数据恢复：即使发生故障，也可以通过重放AOF文件中的命令来恢复数据。

• 适用场景：

  • 当数据不能有任何丢失时。
  • 对于写密集型应用，AOF可以提供更好的数据保护。
  • 需要频繁的写操作，并且要求数据尽可能不丢失的情况下。

• 配置示例：

  • 设置appendfsync选项来控制文件同步的频率，例如：appendfsync everysec 表示每秒同步一次。
  • 定期进行AOF重写以减少文件大小。

1. 组合使用 RDB 和 AOF

在很多情况下会选择同时使用RDB和AOF两种持久化方式，以结合两者的优点：

• 使用RDB进行定期的全量备份，这样即使发生灾难性的故障也能快速恢复。
• 使用AOF来确保数据的完整性和连续性，在服务重启时可以通过重放AOF文件来恢复最新的数据状态。

这种组合方式能够提供较好的性能和数据保护，通常被认为是最佳实践。例如：

• 配置RDB定期生成快照文件，用于灾难恢复。
• 配置AOF以everysec同步策略运行，确保数据的完整性。
• 定期进行AOF重写以保持文件大小在可管理范围内。

3 Redis集群 ​

3.1 面试题 ​

面试题：你们用的是Redis单机还是Redis集群？Redis集群具体怎么做的？

一般在自己学习时用Redis单机，或者一些个人项目中用Redis单机，在生产项目中会使用Redis集群。

Redis单机就是部署一个Redis实例，它不具有高可用，当这个Redis实例挂了将直接影响系统的运行，并且当一台Redis实例不足以承担请求压力时将会影响系统的情况。

Redis集群就是多个Redis实例组成一个集群共同对外提供Redis服务，首先具有高可用性，一个Redis实例挂了还有其它Redis实例对外提供服务，不影响整个集群对外提供服务。还有就是可扩展性，当系统压力比较大时通过扩展Redis实例节点数即可增加集群的服务能力。

所以在生产中正规的项目一般都会使用Redis集群。

那Redis集群具体怎么做的？这个是考察你对Redis集群掌握多少，首先你得知道生产中你的项目用的Redis集群是什么模式，是主从结构、还是分片集群，因为模式不同对于应用程序配置连接Redis的方式也可能不同，其次是对集群之间数据同步、故障转移等基本特性的掌握。至于你的集群有多少Redis节点组成这个一般由运维人员进行部署，作为Java程序员并不清楚生产环境具体节点数是正常的。

所以下一步我们需要搞清楚Redis集群的结构是什么。

3.2 主从结构 ​

3.2.1 介绍 ​

首先说说Redis 的主从结构，下图就是一个简单的Redis主从集群结构：

如图所示，主从集群中有一个master节点、多个slave节点（现在叫replica）组成。

特点：

• 写操作访问master节点，master会自动将数据同步给两个slave节点
• 读操作访问各个slave节点，从而分担并发压力

具体主从集群的搭建请参考“Redis集群搭建”文档，有兴趣的同学在课下可以自已搭建。

3.2.2 主从同步原理 ​

主从集群的master节点是如何把数据同步到slave节点呢？

主从之间通过全量同步、增量同步的方式完成数据同步。

3.2.2.1 全量同步 ​

主从第一次建立连接时，会执行全量同步，将master节点的所有数据都拷贝给slave节点，流程如下：

完整流程描述：

• slave节点请求数据同步
• master节点判断replid，发现不一致表示是第一次同步，进行全量同步

master通过replid判断是否是第一次同步，Replication Id简称replid，是数据集的标记，replid一致则是同一数据集。每个master都有唯一的replid，slave则会继承master节点的replid

• master将完整内存数据生成RDB，发送RDB到slave
• slave清空本地数据，加载master的RDB
• 全量同步完成，后边进行增量同步，命令记录在repl_baklog中，通过offset偏移量master持续将log中的命令发送给slave。

3.2.2.2 增量同步 ​

全量同步需要先做RDB，然后将RDB文件通过网络传输给slave，成本太高了。因此除了第一次做全量同步，其它大多数时候slave与master都是做增量同步。

什么是增量同步？就是只更新slave与master存在差异的部分数据。如图：

那么master怎么知道slave与自己的数据差异在哪里呢?

这就要说到全量同步时的repl_baklog文件了。这个文件是一个固定大小的数组，只不过数组是环形，也就是说角标到达数组末尾后，会再次从0开始读写，这样数组头部的数据就会被覆盖。

repl_baklog中会记录Redis处理过的命令及offset，包括master当前的offset，和slave已经拷贝到的offset：

slave与master的offset之间的差异，就是salve需要增量拷贝的数据了。

随着不断有数据写入，master的offset逐渐变大，slave也不断的拷贝，追赶master的offset：

直到数组被填满：

此时，如果有新的数据写入，就会覆盖数组中的旧数据。不过，旧的数据只要是绿色的，说明是已经被同步到slave的数据，即便被覆盖了也没什么影响。因为未同步的仅仅是红色部分：

但是，如果slave出现网络阻塞，导致master的offset远远超过了slave的offset：

如果master继续写入新数据，master的offset就会覆盖repl_baklog中旧的数据，直到将slave现在的offset也覆盖：

棕色框中的红色部分，就是尚未同步，但是却已经被覆盖的数据。此时如果slave恢复，需要同步，却发现自己的offset都没有了，无法完成增量同步了。只能做全量同步。

repl_baklog大小有上限，写满后会覆盖最早的数据。如果slave断开时间过久，导致尚未备份的数据被覆盖，则无法基于repl_baklog做增量同步，只能再次全量同步。

3.2.3 总结 ​

Redis全量同步和增量同步区别？

• 全量同步：

主从第一次建立连接时，会执行全量同步.

从节点的offset被主的offset覆盖后需要全量同步。

master将完整内存数据生成RDB，发送RDB到slave。

• 增量同步：

slave节点断开又恢复，并且在repl_baklog中能找到offset时执行增量同步。

slave提交自己的offset到master，master获取repl_baklog中从offset之后的命令给slave。

3.3 Redis哨兵 ​

3.3.1 哨兵工作原理 ​

主从结构中master节点的作用非常重要，一旦故障就会导致集群不可用。那么有什么办法能保证主从集群的高可用性呢？

Redis提供了哨兵（Sentinel）机制来监控主从集群监控状态，确保集群的高可用性。

下图是哨兵集群作用原理图：

哨兵的作用如下：

• 状态监控：Sentinel 会不断检查您的master和slave是否按预期

Sentinel 通过定时向master和slave发送ping判断是否下线，如果超过半数的Sentinel 判断master下线则认为是客观下线，按照一定的规则在salve中选择一个作为新的master。

• 故障恢复（failover）：如果master故障，Sentinel会将一个slave提升为master。当故障实例恢复后会成为slave

在所有Sentinel中找一个Leader，由该Leader向新master发送slaveof no one命令，让该节点成为master。

Leader给所有其它slave发送slaveof命令让slave成为新master的slave。

• 状态通知：Sentinel充当Redis客户端的服务发现来源，当集群发生failover时，会将最新集群信息推送给Redis的客户端

3.3.1.1 状态监控 ​

Sentinel基于心跳机制监测服务状态，每隔1秒向集群的每个节点发送ping命令，并通过实例的响应结果来做出判断：

• 主观下线（sdown）：如果某sentinel节点发现某Redis节点未在规定时间响应，则认为该节点主观下线。
• 客观下线(odown)：若超过指定数量（通过quorum设置）的sentinel都认为该节点主观下线，则该节点客观下线。quorum值最好超过Sentinel节点数量的一半，Sentinel节点数量至少3台。

如图：

一旦发现master故障，sentinel需要在salve中选择一个作为新的master，选择依据是这样的：

• 首先会判断slave节点与master节点断开时间长短，如果超过down-after-milliseconds * 10则会排除该slave节点
• 然后判断slave节点的slave-priority值，越小优先级越高，如果是0则永不参与选举（默认都是1）。
• 如果slave-prority一样，则判断slave节点的offset值，越大说明数据越新，优先级越高
• 最后是判断slave节点的run_id大小，越小优先级越高（通过info server可以查看run_id）。

对应的官方文档如下：

问题来了，当选出一个新的master后，该如何实现身份切换呢？

大概分为两步：

• 在多个sentinel中选举一个leader
• 由leader执行failover（故障转移）

3.3.1.2 选举leader ​

首先，Sentinel集群要选出一个执行failover的Sentinel节点，可以成为leader。要成为leader要满足两个条件：

• 最先获得超过半数的投票
• 获得的投票数不小于quorum值

而sentinel投票的原则有两条：

• 优先投票给目前得票最多的
• 如果目前没有任何节点的票，就投给自己

比如有3个sentinel节点，s1、s2、s3，假如s2先投票：

• 此时发现没有任何人在投票，那就投给自己。s2得1票
• 接着s1和s3开始投票，发现目前s2票最多，于是也投给s2，s2得3票
• s2称为leader，开始故障转移

不难看出，谁先投票，谁就会称为leader，那什么时候会触发投票呢？

答案是第一个确认master客观下线的人会立刻发起投票，一定会成为leader。

OK，sentinel找到leader以后，该如何完成failover呢？

3.3.1.3 failover ​

我们举个例子，有一个集群，初始状态下7001为master，7002和7003为slave：

假如master发生故障，slave1当选。则故障转移的流程如下：

1）sentinel给备选的slave1节点发送slaveof no one命令，让该节点成为master

2）sentinel给所有其它slave发送slaveof 192.168.150.101 7002 命令，让这些节点隶属于新master，也就是7002的slave节点，开始从新的master上同步数据。

3）最后，当故障节点恢复后会接收到哨兵信号，执行slaveof 192.168.150.101 7002命令，成为slave：

3.3.2 总结 ​

Redis哨兵的三个作用是什么？

• 集群监控

Redis哨兵每隔1秒向主从节点发送一次ping命令，如果超过一定时间没有相向则认为是主观下线（sdown）

如果大多数Redis哨兵都认为实例主观下线，则判定服务客观下线（odown）

• 故障恢复

首先要在Redis哨兵中选出一个leader，由leader执行failover

• 状态通知

Redis哨兵从slave中选取master后会通过给redis客户端。

Redis哨兵如何判断一个redis实例是否健康？

• Redis哨兵每隔1秒向主从节点发送一次ping命令，如果超过一定时间没有相向则认为是主观下线（sdown）
• 如果大多数Redis哨兵都认为实例主观下线，则判定服务客观下线（odown）

故障转移步骤有哪些？

• 首先要在Redis哨兵中选出一个leader，由leader执行failover
• 选定一个slave作为新的master，执行slaveof no one，切换到master模式
• 然后让所有节点都执行slaveof 新master
• 修改故障节点配置，添加slaveof 新master

Redis哨兵选举leader的依据是什么？

• 票数超过Redis哨兵节点数量1半
• 票数超过设定数量（quorum）
• 一般情况下最先发起failover的节点会当选

Redis哨兵从slave中选取master的依据是什么？

• 首先会判断slave节点与master节点断开时间长短，如果超过down-after-milliseconds * 10则会排除该slave节点
• 然后判断slave节点的slave-priority值，越小优先级越高，如果是0则永不参与选举（默认都是1）。
• 如果slave-prority一样，则判断slave节点的offset值，越大说明数据越新，优先级越高
• 最后是判断slave节点的run_id大小，越小优先级越高（通过info server可以查看run_id）。

3.4 Redis分片集群 ​

3.4.1 介绍 ​

主从模式可以解决高可用、高并发读的问题。但依然有两个问题没有解决：

• 海量数据存储
• 高并发写

要解决这两个问题就需要用到分片集群了。分片的意思，就是把数据拆分存储到不同节点，这样整个集群的存储数据量就更大了。

Redis分片集群的结构如图：

分片集群特征：

• 集群中有多个master，每个master保存不同分片数据 ，解决海量数据存储问题
• 每个master都可以有多个slave节点 ，确保高可用
• master之间通过ping监测彼此健康状态 ，类似哨兵作用
• 客户端请求可以访问集群任意节点，最终都会被转发到数据所在节点

3.4.2 散列插槽 ​

数据要分片存储到不同的Redis节点，肯定需要有分片的依据，这样下次查询的时候才能知道去哪个节点查询。很多数据分片都会采用一致性hash算法。而Redis则是利用散列插槽（hash slot）的方式实现数据分片。

详见官方文档：

在Redis集群中，共有16384个hash slots，集群中的每一个master节点都会分配一定数量的hash slots。具体的分配在集群创建时就已经指定了：

如图中所示：

• Master[0]，本例中就是7001节点，分配到的插槽是0~5460
• Master[1]，本例中就是7002节点，分配到的插槽是5461~10922
• Master[2]，本例中就是7003节点，分配到的插槽是10923~16383

当我们读写数据时，Redis基于CRC16 算法对key做hash运算，得到的结果与16384取余，就计算出了这个key的slot值。然后到slot所在的Redis节点执行读写操作。

不过hash slot的计算也分两种情况：

• 当key中包含{}时，根据{}之间的字符串计算hash slot
• 当key中不包含{}时，则根据整个key字符串计算hash slot

例如：

• key是user，则根据user来计算hash slot
• key是user:{age}，则根据age来计算hash slot

我们来测试一下，先于7001建立连接：

# 进入容器
docker exec -it r1 bash
# 进入redis-cli,这里要加-c表示连接集群
redis-cli -c -p 7001
# 测试
set user jack

结果如下：

可以看到，客户端自动跳转到了5474这个slot所在的7002节点。

现在，我们添加一个新的key，这次加上{}：

# 试一下key中带{}
set user:{age} 21

# 再试一下key中不带{}
set age 20

结果如下：

可以看到user:{age}和age计算出的slot都是741。

3.4.3 故障转移 ​

分片集群的节点之间会互相通过ping的方式做心跳检测，超时未回应的节点会被标记为下线状态。当发现master下线时，会将这个master的某个slave提升为master。

3.4.4 总结 ​

Redis分片集群如何判断某个key应该在哪个实例？

• 将16384个插槽分配到不同的实例
• 根据key计算哈希值，对16384取余
• 余数作为插槽，寻找插槽所在实例即可

如何将同一类数据固定的保存在同一个Redis实例？

• Redis计算key的插槽值时会判断key中是否包含{}，如果有则基于{}内的字符计算插槽
• 数据的key中可以加入{类型}，例如key都以{typeId}为前缀，这样同类型数据计算的插槽一定相同

3.5 总结 ​

面试题：你们用的是Redis单机还是Redis集群？Redis集群具体怎么做的？

面试题：Java程序如何访问Redis集群？

参考："Redis集群搭建"中的“Java客户端连接分片集群”章节。

4 Redis数据类型 ​

4.1 面试题 ​

说一下Redis的常用数据类型？

考察对Redis基本数据类型的掌握，要清楚每种数据类型的特点及常用的操作方法（增、删、查）。

说一下ZSet的底层结构？

4.2 五种数据类型 ​

4.2.1 回顾复习 ​

我们常用的Redis数据类型有5种：String、hash、list、set、zset，还有一些高级数据类型，比如Bitmap、HyperLogLog（简介HLL）、GEO等，其底层都是基于上述5种基本数据类型。因此在Redis的源码中，其实只有5种数据类型。

下边对五种基本数据类型进行总结：

• 字符串(string)：普通字符串，常用

SET key value 设置指定key的值

GET key获取指定key的值

SETEX key seconds value 设置指定key的值，并将 key 的过期时间设为 seconds 秒

SETNX key value 只有在 key 不存在时设置 key 的值

• 哈希(hash)：适合存储对象

HSET key field value 将哈希表 key 中的字段 field 的值设为 value

HGET key field 获取存储在哈希表中指定字段的值

HDEL key field 删除存储在哈希表中的指定字段

HKEYS key 获取哈希表中所有字段

HVALS key 获取哈希表中所有值

HGETALL key 获取在哈希表中指定 key 的所有字段和值

• 列表(list)：按照插入顺序排序，可以有重复元素

LPUSH key value1 [value2] 将一个或多个值插入到列表头部

LRANGE key start stop获取列表指定范围内的元素

RPOP key 移除并获取列表最后一个元素

LLEN key获取列表长度

BRPOP key1 [key2 ] timeout 移出并获取列表的最后一个元素， 如果列表没有元素会阻塞列表直到等待超 时或发现可弹出元素为止

• 集合(set)：无序集合，没有重复元素

SADD key member1 [member2] 向集合添加一个或多个成员

SMEMBERS key 返回集合中的所有成员

SCARD key 获取集合的成员数

SINTER key1 [key2] 返回给定所有集合的交集

SUNION key1 [key2] 返回所有给定集合的并集

SDIFF key1 [key2] 返回给定所有集合的差集

SREM key member1 [member2] 移除集合中一个或多个成员

• 有序集合(sorted set / zset)：集合中每个元素关联一个分数（score），根据分数升序排序，没有重复元素

ZADD key score1 member1 [score2 member2] 向有序集合添加一个或多个成员，或者更新已存在成员的 分数

ZRANGE key start stop [WITHSCORES] 通过索引区间返回有序集合中指定区间内的成员

ZINCRBY key increment member 有序集合中对指定成员的分数加上增量 increment

ZREM key member [member ...]移除有序集合中的一个或多个成员

还需要掌握Redis中的通用命令，主要是针对key进行操作的相关命令：

• KEYS pattern 查找所有符合给定模式( pattern)的 key
• EXISTS key 检查给定 key 是否存在
• TYPE key 返回 key 所储存的值的类型
• TTL key 返回给定 key 的剩余生存时间(TTL, time to live)，以秒为单位
• DEL key 该命令用于在 key 存在是删除 key

4.2.2 总结 ​

说一下Redis的常用数据类型？

• String：存储单个值，适用于缓存和键值存储，常用命令：SET用于设置值，GET用于获取值。
• List：有序、可重复的字符串集合，适用于消息队列和发布/订阅系统，常用命令：LPUSH用于从列表左侧添加元素，LRANGE用于获取指定范围的元素。
• Set：无序、不可重复的字符串集合，适用于标签系统和好友关系等，常用命令：SADD用于向集合添加成员，SMEMBERS用于获取集合所有成员。
• Hash：字段-值对的无序散列，适用于存储对象、缓存和计数器，常用命令：HSET用于设置字段值，HGETALL用于获取散列的所有字段和值。
• Sorted Set：有序的字符串集合，每个成员关联一个分数，适用于排行榜和按分数范围获取成员，常用命令：ZADD用于添加成员及其分数，ZRANGE用于获取指定范围的成员。

4.2 SkipList ​

SkipList（跳表）首先是链表，但与传统链表相比有几点差异：

• 元素按照升序排列存储
• 节点可能包含多个指针，指针跨度不同。

传统链表只有指向前后元素的指针，因此只能顺序依次访问。如果查找的元素在链表中间，查询的效率会比较低。

而SkipList则不同，它内部包含跨度不同的多级指针，可以让我们跳跃查找链表中间的元素，效率非常高。

其结构如图：

我们可以看到1号元素就有指向3、5、10的多个指针，查询时就可以跳跃查找。例如我们要找大小为14的元素，查找的流程是这样的：

• 首先找元素1节点最高级指针，也就是4级指针，起始元素大小为1，指针跨度为9，可以判断出目标元素大小为10。由于14比10大，肯定要从10这个元素向下接着找。
• 找到10这个元素，发现10这个元素的最高级指针跨度为5，判断出目标元素大小为15，大于14，需要判断下级指针
• 10这个元素的2级指针跨度为3，判断出目标元素为13，小于14，因此要基于元素13接着找
• 13这个元素最高级级指针跨度为2，判断出目标元素为15，比14大，需要判断下级指针。
• 13的下级指针跨度为1，因此目标元素是14，刚好于目标一致，找到。

这种多级指针的查询方式就避免了传统链表的逐个遍历导致的查询效率下降问题。在对有序数据做随机查询和排序时效率非常高。

跳表的结构体如下：

typedef struct zskiplist {
    // 头尾节点指针
    struct zskiplistNode *header, *tail;
    // 节点数量
    unsigned long length;
    // 最大的索引层级
    int level;
} zskiplist;

可以看到SkipList主要属性是header和tail，也就是头尾指针，因此它是支持双向遍历的。

跳表中节点的结构体如下：

typedef struct zskiplistNode {
    sds ele; // 节点存储的字符串
    double score;// 节点分数，排序、查找用
    struct zskiplistNode *backward; // 前一个节点指针
    struct zskiplistLevel {
struct zskiplistNode *forward; // 下一个节点指针
unsigned long span; // 索引跨度
    } level[]; // 多级索引数组
} zskiplistNode;

每个节点中都包含ele和score两个属性，其中score是得分，也就是节点排序的依据。ele则是节点存储的字符串数据指针。

4.4 总结 ​

Redis源码中zset(SortedSet)的结构体如下：

typedef struct zset {
    dict *dict; // dict，底层就是HashTable
    zskiplist *zsl; // 跳表
} zset;

zset的底层有两种结果：HashTable、跳表。

Redis的SortedSet底层的数据结构是怎样的？

5.Redis内存回收 ​

5.1 面试题 ​

说一下Redis内存回收机制。

考察对Redis内存过期策略和内存淘汰策略的理解。

Redis之所以性能强，最主要的原因就是基于内存存储。然而单节点的Redis其内存大小不宜过大，会影响持久化或主从同步性能。

我们可以通过修改redis.conf文件，添加下面的配置来配置Redis的最大内存：

maxmemory 1gb

当内存达到上限，就无法存储更多数据了。因此，Redis内部会有两套内存回收的策略：

• 内存过期策略
• 内存淘汰策略

5.2.内存过期处理 ​

存入Redis中的数据可以配置过期时间，到期后再次访问会发现这些数据都不存在了，也就是被过期清理了。

5.2.1.过期命令 ​

Redis中通过expire命令可以给KEY设置TTL（过期时间），例如：

# 写入一条数据
set num 123
# 设置20秒过期时间
expire num 20

不过set命令本身也可以支持过期时间的设置：

# 写入一条数据并设置20s过期时间
set num EX 20

当过期时间到了以后，再去查询数据，会发现数据已经不存在。

5.2.2.过期策略 ​

那么问题来了：

• Redis如何判断一个KEY是否过期呢？
• Redis又是何时删除过期KEY的呢？

Redis是何时删除过期KEY的呢？

Redis并不会在KEY过期时立刻删除KEY，因为要实现这样的效果就必须给每一个过期的KEY设置时钟，并监控这些KEY的过期状态。无论对CPU还是内存都会带来极大的负担。

Redis的过期KEY删除策略有两种：

• 惰性删除
• 周期删除

惰性删除：

当一个客户端尝试访问一个键时，Redis 会检查该键是否已经过期。如果过期，Redis 将删除该键，并返回一个表示键不存在的响应给客户端。这种方式确保了对内存的有效管理，但可能在高并发访问过期键的情况下导致性能下降。

定期删除：

Redis 还有一个后台线程，以一定的频率检查过期的键并删除它们。这个频率可以通过 server.hz 参数配置。server.hz 定义了 Redis 服务器每秒运行维护任务的次数，包括但不限于过期键的清理。默认情况下，server.hz 的值为 10，意味着每秒进行 10 次检查。

如何调整 server.hz？

• 提高频率：如果您的应用中有很多短生命周期的键，并且希望更快地回收这些键所占用的内存，可以考虑增加 server.hz 的值。但是请注意，更高的 server.hz 会增加 CPU 的使用率，因为它会导致更频繁的后台任务执行。
• 降低频率：如果您发现 Redis 服务器的 CPU 使用率较高，而您的应用对过期键的清理速度要求不是特别严格，可以考虑降低 server.hz 的值以减少 CPU 负载。

配置方法：

要修改 server.hz 的值，您可以在 Redis 的配置文件 redis.conf 中找到相应的设置，并根据需要更改它。

5.3.内存淘汰策略 ​

对于某些特别依赖于Redis的项目而言，仅仅依靠过期KEY清理是不够的，内存可能很快就达到上限。因此Redis允许设置内存告警阈值，当内存使用达到阈值时就会主动挑选部分KEY删除以释放更多内存。这叫做内存淘汰机制。

Redis支持8种不同的内存淘汰策略：

1）noeviction： 不删除，直接返回报错信息。

2）volatile-lfu：在设置了过期时间的key中，移除最近最少（最少频率使用）使用的key。

4）volatile-lru：在设置了过期时间的key中，移除最近最久未使用的key。

3）volatile-ttl： 在设置了过期时间的key中，移除准备过期的key。

5）volatile-random：在设置了过期时间的key中，随机移除某个key。

6）allkeys-random：随机移除某个key。

7）allkeys-lru：移除最久未使用的key。

8）allkeys-lfu：移除最近最少使用的key。

比较容易混淆的有两个算法：

• LFU（Least Frequently Used），最少频率使用。会统计每个key的访问频率，值越小淘汰优先级越高。
• LRU（Least Recently Used），最近最久未使用。用当前时间减去最后一次访问时间，这个值越大则淘汰优先级越高。

不过这里大家要注意一下：Redis中的KEY可能有数百万甚至更多，每个KEY都有自己访问时间或者逻辑访问次数。我们要找出时间最早的或者访问次数最小的，难道要把Redis中所有数据排序？

要知道Redis的内存淘汰是在每次执行命令时处理的。如果每次执行命令都先对全量数据做内存排序，那命令的执行时长肯定会非常长，这是不现实的。

所以Redis采取的是抽样法，即每次抽样一定数量（maxmemory_smples）的key，然后基于内存策略做排序，找出淘汰优先级最高的，删除这个key。这就导致Redis的算法并不是真正的LRU，而是一种基于抽样的近似LRU算法。

5.4.总结 ​

Redis何时删除过期KEY？如何删除？/删除策略？

Redis淘汰策略有哪几种？