redis应用及原理

视频：https://www.bilibili.com/video/BV1cr4y1671t

链接https://pan.baidu.com/s/1189u6u4icQYHg_9_7ovWmA?pwd=eh11
提取码：eh11

社交类项目

短信登陆
商户查询缓存：缓存雪崩穿透击穿
博客
优惠券秒杀：计算器 lua 分布式锁队列
好友关注：set集合

项目搭建

前后端分离

start nginx.exe

数据库表

user
user_info
shop
shop_type
blog
follow
voucher
voucher_order

public class UserHolder {
    private static final ThreadLocal<UserDTO> tl = new ThreadLocal<>();
    public static void saveUser(UserDTO user){
        tl.set(user);
    }
    public static UserDTO getUser(){
        return tl.get();
    }
    public static void removeUser(){
        tl.remove();
    }
}

去除敏感信息：创建一个小的类，然后copy信息

1 2	`// 把1中有的都给2，并创建对象 BeanUtil.copyProperties(user,UserDTO.class))`

redis：验证码、用户信息都存入redis中。

用户信息用hash（BeanUtil.beanToMap），添加有效期，登录添加登陆后更新有效期
拦截器类没有交给spring，不能注入redis，因此添加一个成员变量
StringRedisTemplate 需要保证hash中key value 都是string。所以beantomap时需要对类型进行转化
问题：有的请求没有拦截，导致没有刷新。再添加一个。一个全部刷新添加、一个拦截

商户查询缓存

缓存：浏览器缓存 redis缓存数据库缓存索引

问题：数据一致性

shop、shop_type添加缓存

缓存更新策略

低一致性的数据：内存淘汰
高一致性：主动更新+超时

主动更新

更新数据库的同时更新缓存
- 更新还是删除缓存：删除比较一致性好。防止多次更新
- 如何保证同时成功失败：单体：事务；分布式：TCC分布式事务
- 先更新数据库还是缓存？都可能有问题
  - 先删缓存：A删完缓存来了查询B，B查询完成后写入脏数据到redis
  - 先数据库：出问题概率更低（B来的时候没有缓存，B读取数据库，A更新数据并删除redis，B写脏数据到redis），但A写数据库的过程，B查的都是旧数据
缓存、数据库整合成一个服务
先更新缓存、异步更新到数据库

所以：查询商品：添加超时，并且更新后删除缓存

此外：还可以直接加读写锁，更新时直接加写锁。读取时读redis加读锁，读数据库并更新加写锁并发251集

ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

public int update(String sql, Object... params) {
    SqlPair key = new SqlPair(sql, params);
    // 加写锁, 防止其它线程对缓存读取和更改
    lock.writeLock().lock();
    try {
        int rows = genericDao.update(sql, params);
        map.clear();
        return rows;
    } finally {
        lock.writeLock().unlock();
    }
}

public T queryOne(Class<T> beanClass, String sql, Object... params) {
    SqlPair key = new SqlPair(sql, params);
    // 加读锁, 防止其它线程对缓存更改
    lock.readLock().lock();
    try {
        T value = map.get(key);
        if (value != null) {
            return value;
        }
    } finally {
        lock.readLock().unlock();
    }
    // 加写锁, 防止其它线程对缓存读取和更改
    lock.writeLock().lock();
    try {
        // get 方法上面部分是可能多个线程进来的, 可能已经向缓存填充了数据
        // 为防止重复查询数据库, 再次验证
        T value = map.get(key);
        if (value == null) {
            // 如果没有, 查询数据库
            value = genericDao.queryOne(beanClass, sql, params);
            map.put(key, value);
        }
        return value;
    } finally {
        lock.writeLock().unlock();
    }

}

缓存穿透

Redis和数据库都不存在的数据，不停的请求每次都会打到数据库上。

缓存空对象：直接缓存一个空值“”，需要添加TTL。可能照成短期的不一致。代码写法
布隆过滤器：添加一层判断是否纯在。内存少但实现复杂

此外还有：数据格式校验、用户权限校验、热点参数限流 cloud

缓存雪崩

同时大量缓存失效或者宕机。

TTL可以添加随机
Redis集群
降级限流确保数据库安全
多级缓存

缓存击穿

查询数据库但在没有构建出缓存的这段时间，同时大量重建 击穿了数据库。

热点key（高并发），并且重新构建业务比较复杂（查询数据库的时间很长，这个时间大家都来查了）。

互斥锁：加一个查询锁。没有锁的休息下重试。可能死锁 （锁的可能会有问题：加版本加lua脚本见秒杀.md）
逻辑过期：永不过期防止查不到，并且添加一个逻辑过期时间字段，如果逻辑过期了，就返回旧值并异步更新一下（锁）

互斥锁

setIfAbsent实现

private boolean isLock(String key){
    Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(key, "1");
    return BooleanUtil.isTrue(aBoolean);
}
private boolean unLock(String key){
    Boolean delete = stringRedisTemplate.delete(key);
    return delete;
}

public Shop queryWithMutex(long id){
    String s = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    if (s != null){
        if (s.isEmpty()){
            return null;
        }
        return JSONUtil.toBean(s, Shop.class);
    }
    Shop shop = null;
    String lockKey = "lock:shop:" + id;
    try {
        if(!isLock(lockKey)){
            Thread.sleep(50);
            queryWithMutex(id);
        }
        // double check 没有的话也会访问多次数据库
        s = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
        if (s != null){
            if (s.isEmpty()){
                return null;
            }
            return JSONUtil.toBean(s, Shop.class);
        }
        // 模拟
        shop = getById(id);
        Thread.sleep(200);

        if (shop == null){
            stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, "", CACHE_NULL_TTL, TimeUnit.MINUTES );
            return null;
        }
        stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES );
    }
    catch (InterruptedException e) {
        e.printStackTrace();
    }finally {
        unLock(lockKey);
    }
    return shop;
}

逻辑过期

RedisData: 包装类保存时间和数据

public class RedisData {
    private LocalDateTime expireTime;
    private Object data;
}

// 取出
RedisData redisData = JSONUtil.toBean(s, RedisData.class);
JSONObject data = (JSONObject)redisData.getData();
Shop shop = JSONUtil.toBean(data, Shop.class);

先预热下数据,该数据逻辑上过期了
redis没有直接返回
有再看下时间要不要重建
最后修改下数据库，1m发起1000次请求，重建延时为200ms，可以看到后面的数据可以查到真的，前面是脏的

没有doublecheck但数据库依旧只查询了一次，原因未知

// 使用逻辑过期
private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);
public Shop queryWithLogicExpire(long id){
    String s = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    if (s == null || s.isEmpty()){
        return null;
    }

    RedisData redisData = JSONUtil.toBean(s, RedisData.class);
    JSONObject data = (JSONObject)redisData.getData();
    Shop shop = JSONUtil.toBean(data, Shop.class);
    LocalDateTime expireTime = redisData.getExpireTime();
    if (expireTime.isAfter(LocalDateTime.now())){
        return shop;
    }

    String lockKey = LOCK_SHOP_KEY + id;
    // 没拿到锁直接返回
    if (!isLock(lockKey)) {
        return shop;
    }

    // double check 但没发现影响
    //        s = stringRedisTemplate.opsForValue().get(CACHE_SHOP_KEY + id);
    //        if (s == null || s.isEmpty()){
    //            return null;
    //        }
    //        redisData = JSONUtil.toBean(s, RedisData.class);
    //        data = (JSONObject)redisData.getData();
    //        shop = JSONUtil.toBean(data, Shop.class);
    //        expireTime = redisData.getExpireTime();
    //        if (expireTime.isAfter(LocalDateTime.now())){
    //            return shop;
    //        }

    CACHE_REBUILD_EXECUTOR.submit(() -> {
        try {
            saveShop2Redis(id, 20L);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException();
        } finally {
            unLock(lockKey);
        }
    });

        return shop;

    }

工具类

对象->string类型 TTL
对象->string类型逻辑TTL
queryWithPassThrough 空字符串防止缓存穿透
- pre id Class -> 对象 null防止缓存穿透，Function<ID, R> dbFallback, Long time, TimeUnit unit重建
queryWithLogicalExpire
- pre id Class -> 对象 Function<ID, R> dbFallback, Long time, TimeUnit unit重建
queryWithMutex

秒杀

代金券就是商品

tb_voucher_order：订单表

ID生成器

id没有使用自增：规律性明显、表拆分后自增会冲突

全局id生成器：唯一、高可用（集群）、高性能、递增、安全

时间戳用的秒，序列号为一天的自增长（需要一个prefix）：方便统计计量、控制大小

LocalDateTime.now(); // 时间
now.toEpochSecond(ZoneOffset.UTC); // 转为秒
now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd")); // 转为特定格式字符串


CountDownLatch 可以实现线程的等待
CountDownLatch(300);// 计数
countDownLatch.countDown();
countDownLatch.await();// 等待减完 阻塞

优惠券下单

同样两个表，普通表和秒杀表

普通表有基本信息（type区分普通还是秒杀），秒杀表扩展 库存时间。voucher类中有额外字段，用@TableField(exist = false)标识。

添加优惠券：一个接口同时添加两个表。

超卖

悲观锁：直接一个redis.setnx锁
乐观锁：更新数据时，成功率很低
- 添加版本号，修改前需要查版本号 set stock = stock -1 where version = initVersion
- CAS(compare and set)：库存本来就是版本号，where stock = initStock ，但这样会导致成功率低； ==修改为==：where stock > 0 也就是秒杀项目的基础用法

一人一单

唯一索引
加锁: 单机情况
1. sychronized：锁定函数,阻塞的
2. 缩小范围，提取一人一单开始锁定，并且通过userid标识 synchronized (userId.toString().intern())
3. 小函数添加事务，提取出真的的代理对象调用小函数。引入依赖、添加注解

synchronized (userId.toString().intern()) {
    // 获取代理对象(事务)
    IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy() ;
    return proxy.createVoucherOrder(voucherId);
}

问题：微服务项目，不同服务中的普通线程锁就失败了。无法跨进程

分布式锁

普通set del => 加超时 => 加版本 => 加Lua

set lock thread1 NX EX 10

添加SimpleRedisLock类专门处理锁，需要传入name 和 redis

@Override
    public boolean tryLock(long timeoutSec) {
        // 获取线程标示
        String threadId = ID_PREFIX + Thread.currentThread().getId();
        // 获取锁
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(success);
    }

    @Override
    public void unlock() {
        // 调用lua脚本
        stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX + Thread.currentThread().getId());
    }

	// 非原子
    public void unlock2() {
        // 获取线程标示
        String threadId = ID_PREFIX + Thread.currentThread().getId();
        // 获取锁中的标示
        String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
        // 判断标示是否一致
        if(threadId.equals(id)) {
            // 释放锁
            stringRedisTemplate.delete(KEY_PREFIX + name);
        }
    }

lua

数组下标从1开始，需要指明key的数量

local key = KEYS[1]
local value = ARGV[1]
redis.call('SET', key, value)
return 'OK'

eval "local key = KEYS[1]\nlocal value = ARGV[1]\nredis.call('SET', key, value)\nreturn 'OK'" 1 mykey hello


unlock.lua：
 if redis.call("get",KEYS[1])==ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

脚本定义和调用

UNLOCK_SCRIPT = new DefaultRedisScript<>();
UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
UNLOCK_SCRIPT.setResultType(Long.class);

stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX + Thread.currentThread().getId());
                
                
UNLOCK_SCRIPT也可以直接交给spring

存在问题

不可重用 A拿了锁调用B，B也需要锁
不可重试，只try一次
超时释放和业务时间冲突
主从一致性，主节点宕机后存在还能没同步到从节点

Redisson

基本使用

1.首先，在 pom.xml 文件中添加 Redisson 的依赖：

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.16.1</version>
</dependency>

2.在 Spring Boot 应用中，使用 Redisson 的方式通常是通过创建一个 RedissonClient 实例与 Redis 进行通信。RedissonClient 可以根据 Redis 的不同部署方式来进行配置。下面是一个基于单节点的 Redisson 配置示例：

@Configuration
public class RedissonConfig {
    @Bean(destroyMethod = "shutdown")
    public RedissonClient redisson() {
        Config config = new Config();
        config.useSingleServer()
                .setAddress("redis://localhost:6379")
                .setDatabase(0)
                .setPassword("password");
        return Redisson.create(config);
    }
}

3.执行

@Autowired
private RedissonClient redisson;

void doSomething() {
    RLock lock = redisson.getLock("lock:order:"+userId);
    lock.tryLock(1, 10, TimeUnit.SECONDS); // 阻塞尝试1s，过期时间10s
    try {
        // 被锁保护的代码区
    } finally {
        lock.unlock();
    }
}

可重用

Redisson自带可重用，可以打断点看redis中

实现原理：添加一个计数，如果是线程值相同，那么说明是同一个线程，可以获取锁，并数量+1

hash结构实现”： lock:order ( uuid+thread : count)

释放锁时，先减，减到零再释放锁

重试

获取锁成功直接返回，失败会返回过期时间
获取锁失败，订阅别人的释放tread信号
等到后，再去while true尝试锁，通过信号量

主从一致性

开多台主机，重用保证主机的可靠性。相当于把redis主从机都做了复制

异步

阻塞队列

jvm中的服务，受到本地性能影响；并且数据可能丢失

添加任务：往队列中add
一个BlockQueue以及一个线程池，线程池中一个线程任务为不停的从队列中拿出任务并执行，这样就实现了异步

消息队列

List：List模拟消息队列，单消费者
1
通过BRPOP 实现阻塞的提取任务
PubSub：点对点的消息队列，多生产多个消费者，可以实现通过通配符实现订阅
1
2
3
4
subscribe(channel) psubscribe(pattern) publish(channel)
消息堆积有上限，消息可能丢失（如果没有人接受的）

Stream：比较完善的模型，是一种新的数据类型

XADD key k v
# 创建名为users的队列，并向其中发送一个消息，内容是: {name= jack , age=21},并且使用Redis自动生成ID
	XADD users * name jack age 21

XREAD COUNT 1 STREAM key 0(从哪个消息开始，$为最新)   BLOCK 阻塞   读完不删除

消费者组：一个组有多个消费者，共同消费；消息存在标识代表是否处理过，已读未确认的加入pending

XACK s1 g1 id

分布式缓存

数据丢失主从持久化
并发能力读写分离
故障恢复哨兵检测监控状态
存储能力单节点数据太大了，分片集群

持久化

为了安全性使用AOF（1s刷盘），RDB（频率不高、空间小）定期手动进行备份

RDB

RDB Redis Database Backup file 快照（snapshot）持久化方式
保存到磁盘,如果docker启动还需要挂载

save   # 阻塞  用于关机
bgsave # 异步执行 子进程

save 300 10   # 100s内有10个key修改了则触发bgsave
save "" # 禁用

dbfilename dump.rdb # 文件名
dir ./7001  #   相对于 工作目录 的路径

bgsave中fork过程是需要耗时的，需要拷贝内存。实际上是拷贝页表。

如果在save过程写请求进来，会修改一个副本内存

AOF

(Append-Only File) 记录下命令，优先使用AOF恢复

appendonly yes
appendfilename "appendonly.aof"

appendfsync always # 立即磁盘
appendfsync everysec # 先写入缓存，1s写回磁盘默认 
appendfsync no  # 先写入缓存，操作系统写回

同样会在开始前加载，结束后保存

但：同一个key实际上前面的操作没有意义，bgrewriteaof 异步只保留最后一次

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

主从

搭建

开启RDB 关闭AOF
拷贝三个配置文件，需要不同的port和dir文件路径 master关闭bind

slave
    SLAVEOF masterip masterport 当前成为成为slave     配置文件或者cli执行命令

master
	info replication 显示信息

至此master可读可写，slave只可以读，master写时会同步到从

原理

基于RDB

第一次判断：master有唯一replid，是数据集的标记，不一致说明是第一次
是否落后：offset 偏移量代表数据，offset不一致说明数据落后了

全量同步

第一次：全量同步；如果尝试做增量同步失败后

增量同步

根据offset，发回repl_backlog差距
log是一个环形数组，如果差距写满了，就需要全量同步

其他优化

repl-diskless-sync yes # 直接网络发送而不是先磁盘
适当提高repl_backlog大小
主从从结构

故障恢复-哨兵

哨兵也是一个小集群

作用和原理

master宕机，立马更换一个
1. 与master断开时间小于配置
2. 再根据优先级
3. 最后判断offset
slave宕机重启
如果发生了更换，客户端放如何知道？通知客服端

心跳监控

迁移master

备选的变成master
通知其他slave
原来master配置文件改为slaveof new

搭建

sentinel.conf

启动 redis-sentinel path.conf

感知切换

redistemplate基于lettuce实现感知自动切换

不需要配置redis地址，而是sentinel地址

分片集群

前面的问题：1.海量数据存储 2.高并发的写操作

多个master 保存不同的数据通过插槽实现；master之间可以同时执行命令，高性能
master之间相互检测健康不需要sentinel；高可用
访问集群的任意节点，会自动转发到正确节点

配置

1.启动 master slave所有的都一样, config-file每个人都有自己的当前启动后相互还没有关系

2.创建集群

redis-cli --cluster create --cluster-replicas 2 ip1 ip2 ....

1	`redis-cli --cluster create --cluster-replicas 1 127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 127.0.0.1:7006`

一个master对应两个slave，前面的ip为master

查看集群状态：redis-cli -p 7001 cluster nodes

连接：redis-cli -c -p 7001

散列插槽

每一个master都会散列到0~16383个插槽上的一部分

key的有效部分{hash_tag}（控制某一类在一个redis上）也会hash计算插槽%16384，得到插槽位置：数据和插槽绑定

集群伸缩

redis-cli --cluster help

redis-cli --cluster add-node ip:port 集群ip:port

分配插槽

redis-cli --cluster reshard ip:port  
3000 # 移动多少个
to的id
from的id
done

故障转移

当一个master宕机？

集群可以实现自动切换,master宕机换下一个slave，slave也宕机插槽就宕机了

手动转移：进入到slave CLUSTER FAILOVER

代码

多级缓存

最佳实践

键值设计

key

bigkey

合适数据结构

存储对象：

json字符串方便
直接打散，一个字段一个key value 浪费空间
hash；entry不超过1000 如果太大拆分取模成多个hash

批处理优化

Pipeline

单次：网络延时(ms级)+执行延时(us级) 一次执行多条指令，如mset，但收到指令集的限制

pipeline：不受指令限制，一次发送多条指令，不具备原子性

问题：在集群下，多个命令（pipeline mset）必须要在同一个插槽，才能映射到一台机器上

集群下的批处理

mset默认情况下如果key在不同节点上，会报错

集群下，spring mset就会默认按照插槽划分，并且创建线程异步并行执行（方法三）

服务端优化

持久化配置

缓存的redis实例开启一个单独实例，不需要做持久化
为了安全性使用AOF，RDB定期手动进行备份

慢查询

单线程任务，慢查询会影响其他命令

1 2	`slowlog-log-slower-than 默认10000us 建议1000 慢查询会放入日志中 slowlog-max-len 日志长度`

SLOWLOG GET <n> 命令来获取慢查询日志中的最近 N 条记录，或使用 SLOWLOG RESET 命令清空慢查询日志

命令及安全配置

在没有密码并且root启动redis时，修改rbd文件指向ssh文件，并设置value为公钥即可

上线时禁用flushdb， config set 等命令，通过rename command
bind绑定局域网网卡
开启防火墙
不要使用root启动
非默认端口，设置密码

内存配置

MEMORY STATS 命令将返回以下统计数据：

peak.allocated：Redis 运行期间分配的内存峰值。
total.allocated：Redis 运行期间分配的总内存量。
startup.allocated：Redis 启动时分配的内存量。
replication.backlog：复制积压缓冲区使用的内存量。
clients.slaves：连接的从节点数量。
clients.normal：连接的普通客户端数量。
aof.buffer：AOF 缓冲区使用的内存量。
lua.caches：Lua 脚本缓存使用的内存量。
overhead.total：Redis 内部开销占用的内存量。
dataset.bytes：数据集占用的内存量。
dataset.keys：数据集中键的数量。

集群 or 单体

当部分插槽不可用时（master slave都宕机），默认集群宕机

cluster-require-full-coverage yes

节点中会互相ping监控状态，如果集群数量太多，ping也需要大量的带宽

原理

数据结构

SDS

sds.h
    SDS_TYPE_8、SDS_TYPE_16、SDS_TYPE_32、SDS_TYPE_64  节约内存
    
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[];
};

可以避免缓冲区溢出：C 语言中的字符串被修改（比如拼接）时，一旦没有分配足够长度的内存空间，就会造成缓冲区溢出。SDS 被修改时，会先根据 len 属性检查空间大小是否满足要求，如果不满足，则先扩展至所需大小再进行修改操作。
获取字符串长度的复杂度较低：C 语言中的字符串的长度通常是经过遍历计数来实现的，时间复杂度为 O(n)。SDS 的长度获取直接读取 len 属性即可，时间复杂度为 O(1)。
减少内存分配次数：为了避免修改（增加/减少）字符串时，每次都需要重新分配内存（C 语言的字符串是这样的），SDS 实现了空间预分配和惰性空间释放两种优化策略。当 SDS 需要增加字符串时，Redis 会为 SDS 分配好内存，并且根据特定的算法分配多余的内存，这样可以减少连续执行字符串增长操作所需的内存重分配次数。当 SDS 需要减少字符串时，这部分内存不会立即被回收，会被记录下来，等待后续使用（支持手动释放，有对应的 API）。
二进制安全：C 语言中的字符串以空字符 \0 作为字符串结束的标识，这存在一些问题，像一些二进制文件（比如图片、视频、音频）就可能包括空字符，C 字符串无法正确保存。SDS 使用 len 属性判断字符串是否结束，不存在这个问题。

动态扩容

预分配多一点内存可以减少分配次数

IntSet

typedef struct intset {
uint32_t encoding; /* 编码方式，支持存放16位、32位、64位整数*/ 升级机制节约内存
uint32_t Length; /*元素个数*/
int8_t contents[]; /*整数数组，保存集合数据   实际大小由encoding决定 其实就是一个起点指针*/
} intset;

为了方便查找，会有序、唯一存放（二分）。根据所占字节以及index索引 start + index * encoding

如果存放的数超过了范围，会升级。需要将元素移动到合适位置

适合数据量不大Set，有序方便查找某一个元素是否在集合中

Dict

key-value pair通过Dict实现，实现原理和java的hashmap很像，同样是2^n^大小的数组+链表

扩容时触发rehash

两个ht用于扩容时存放新的hash表

渐进扩容

除了扩容，此外如果LoadFactor < 0.1会收缩

将空间申请到dict.ht[1]，此外标记rehashidx=0代表迁移工作开始

渐进rehash：每次执行新增、查询、修改、删除操作时，都检查一下dict.rehashidx是否大于-1, 如果是则将dict.ht[0].table[rehashidx]的entry链表rehash到dict.ht[1]（需要重新计算hash），并且将rehashidx++。直至dict.ht[0]的所有数据都rehash到dict.ht[1]

如果rehash在进行中，删改查需要在两个table中都去找，增加直接到dict.ht[1]

最后将ht[0]指向ht[1]

ZipList

连续空间（所以不能太大；4~64kb）的双端链表，节省内存和提供快速的顺序访问

指针也要占内存（8B），如何节约？通过分配连续内存，并记录下前一个entry的长度实现前遍历，当前长度通过encoding记录。更少的内存碎片
不能随机读取
适用于顺序读取(或两端读取)操作频繁，写入操作较少 以及 小型集合（数小、数量少）

entry：previous用于向前遍历，encoding记录当前长度

encoding

连锁更新问题

连续保存多个253B数据：长度保存字段1B或5B，如果前一个过长当前就要变成5B，导致当前entry超过254连锁更新下一个节点

QuickList

ZipList需要连续空间，找不到怎么办？

QuickList链表连接多个ZipList，兼具链表（非连续空间）数组（节约内存）的优点

list-max-ziplist-size：限制ziplist的entry数量或内存大小
list-compress-depth：是否压缩中间节点以及压缩的深度

SkipList

QuickList需要全部遍历，如何实现LogN随机查询？

顺序存储
包含多个指针，指针的跨度不同（最多32级）
查找过程向右或向下走

每一个节点有一个下一个节点数组，不同节点数组长度不同

RedisObject

Redis中的键值都会被封装为RedisObject，头需要16B

数据结构

数据量小的时候，尽量节约空间；数据量大不得不空间换时间

String

其基本编码方式是RAW,基于简单动态字符串(SDS)实现，存储上限为512mb。
如果存储的SDS长度小于44字节，则会采用EMBSTR编码,此时object head与SDS是一段连续空间。申请内存时只
需要调用1次内存分配函数,效率更高。（小于44字节，一共分配小于64）
如果存储的字符串是整数值,并且大小在LONG_ MAX范围内，则会采用INT编码:直接将数据保存在RedisObject的
ptr指针位置(刚好8字节) ,不再需要SDS了。

List

LinkedList :普通链表，可以从双端访问，内存占用较高，内存碎片较多 3.2之前
ZipList :压缩列表，可以从双端访问，内存占用低，存储上限低 3.2之前
QuickList 3.2之后

set

唯一但不有序
需要求交集、并集、差集

需要高效判断元素是否存在：Hash， key来存储元素，value=null。如果都是int，会使用intset

robj *setTypeCreate(sds value) {
    //判断value是否是数值类型long long
    if (i sSdsRepresentableAsLongLong(value,NULL) == C. _0K)
        //如果是数值类型，则采用IntSet编码
        return createIntsetobject() ;
    //否则采用默认编码，也就是HT
    return createSetObject();
}

Zset

元素少直接用zipList，两个entry 分别保存element和score；score小的在前面

Hash

Hash结构默认采用ZipList编码，用以节省内存。ZipList中相邻的两个entry 分别保存field和value
当数据量较大时，Hash结构会转为HT编码，也就是Dict, 触发条件有两个:
- ZipList中的元素数量超过了hash-max-ziplist-entries (默认512)
- ZipList中的任意entry大小超过了hash-max-ziplist-value ( 默认64字节)