无标签

Redis

发布日期: 2023-10-13

Redis高级篇-黑马

分布式缓存

当然分片集群也内置了哨兵（各个master节点）来进行故障恢复

持久化

RDB全称Redis Database Backup file（Redis数据备份文件），简单来说就是把内存中的所有数据都记录到磁盘中。快照文件称为RDB文件，默认是保存在当前运行目录。

RDB持久化在四种情况下会执行：

执行save命令
执行bgsave命令
Redis停机时
触发RDB条件时

save命令，主进程执行RDB,这个过程中其它所有命令都会被阻塞。只有在数据迁移时可能用到。

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

bgsave开始时会fork主进程得到子进程，子进程共享主进程的内存数据。完成fork后读取内存数据并写入 RDB 文件。
fork采用的是copy-on-write技术：
当主进程执行读操作时，访问共享内存；
当主进程执行写操作时，则会拷贝一份数据，执行写操作。

Redis停机时会执行一次save命令，实现RDB持久化。

Redis内部有触发RDB的机制，可以在redis.conf文件中找到。比如，900秒内，如果至少有1个key被修改，则执行bgsave。

AOF全称为Append Only File（追加文件）。Redis处理的每一个写命令都会记录在AOF文件，可以看做是命令日志文件。

AOF的刷盘，同步刷盘太耗费性能，交给操作系统则可靠性差，则推荐everysec每秒刷盘

通过执行bgrewriteaof命令，可以让AOF文件执行重写功能，用最少的命令达到相同效果

主从

假设有A、B两个Redis实例，如何让B作为A的slave节点？
在B节点执行命令：slaveof A的IP A的port

第一次是全量同步，1.1请求数据同步

master判断一个节点是否是第一次同步的依据，就是看replid是否一致。

之后几次一般都是增量同步，除非差距过大

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

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

主从优化：可以从以下几个方面来优化Redis主从就集群：

在master中配置repl-diskless-sync yes启用无磁盘复制，避免全量同步时的磁盘IO。
Redis单节点上的内存占用不要太大，减少RDB导致的过多磁盘IO
适当提高repl_baklog的大小，发现slave宕机时尽快实现故障恢复，尽可能避免全量同步
限制一个master上的slave节点数量，如果实在是太多slave，则可以采用主-从-从链式结构，减少master压力

主从从架构图：

哨兵（Sentinel）

哨兵的作用如下：

监控：Sentinel 会不断检查您的master和slave是否按预期工作
自动故障恢复：如果master故障，Sentinel会将一个slave提升为master。当故障实例恢复后也以新的master为主
通知：Sentinel充当Redis客户端的服务发现来源，当集群发生故障转移时，会将最新信息推送给Redis的客户端

Sentinel基于心跳机制监测服务状态，每隔1秒向集群的每个实例发送ping命令：

一个sentinel判断是主观下线，过半sentinel判断则构成客观下线

故障恢复：

首先选定一个slave作为新的master，执行slaveof no one
然后让所有节点都执行slaveof 新master
修改故障节点，执行slaveof 新master

配置读写分离：

Spring的RedisTemplate底层利用lettuce实现了节点的感知和自动切换。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

然后在配置文件application.yml中指定redis的sentinel相关信息：

spring:
  redis:
    sentinel:
      master: mymaster
      nodes:
        - 192.168.150.101:27001
        - 192.168.150.101:27002
        - 192.168.150.101:27003

配置读写分离

在项目的启动类中，添加一个新的bean：

@Bean
public LettuceClientConfigurationBuilderCustomizer clientConfigurationBuilderCustomizer(){
    return clientConfigurationBuilder -> clientConfigurationBuilder.readFrom(ReadFrom.REPLICA_PREFERRED);
}

这个bean中配置的就是读写策略，包括四种：

MASTER：从主节点读取
MASTER_PREFERRED：优先从master节点读取，master不可用才读取replica
REPLICA：从slave（replica）节点读取
REPLICA _PREFERRED：优先从slave（replica）节点读取，所有的slave都不可用才读取master

分片集群

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

海量数据存储问题
高并发写的问题

使用分片集群可以解决上述问题，如图:

分片集群特征：

集群中有多个master，每个master保存不同数据
每个master都可以有多个slave节点
master之间通过ping监测彼此健康状态
客户端请求可以访问集群任意节点，最终都会被转发到正确节点

插槽：存储数据是存在插槽上，0~16383共16384个插槽（hash slot）上，然后根据master数量再进行分区域。

数据key不是与节点绑定，而是与插槽绑定。redis会根据key的有效部分计算插槽值，分两种情况：

key中包含”{}”，且“{}”中至少包含1个字符，“{}”中的部分是有效部分
key中不包含“{}”，整个key都是有效部分

例如：key是num，那么就根据num计算，如果是{itcast}num，则根据itcast计算。计算方式是利用CRC16算法得到一个hash值，然后对16384取余，得到的结果就是slot值。

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

这一类数据使用相同的有效部分，例如key都以{typeId}为前缀

集群可以进行伸缩，要执行一些redis cluster指令，还要分配插槽。

RedisTemplate底层同样基于lettuce实现了分片集群的支持，而使用的步骤与哨兵模式基本一致：

多级缓存

在tomcat那的进程缓存，可以使用Caffeine

在Nginx中也是写lua，可以使用OpenResty(OpenResty® 是一个基于 Nginx的高性能 Web 平台)

数据同步-canal

Canal就是把自己伪装成MySQL的一个slave节点，从而监听master的binary log变化。再把得到的变化信息通知给Canal的客户端，进而完成对其它数据库的同步。

Canal是基于mysql的主从同步来实现的，MySQL主从同步的原理如下：

1）MySQL master 将数据变更写入二进制日志( binary log），其中记录的数据叫做binary log events
2）MySQL slave 将 master 的 binary log events拷贝到它的中继日志(relay log)
3）MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

引入依赖：

<dependency>
    <groupId>top.javatool</groupId>
    <artifactId>canal-spring-boot-starter</artifactId>
    <version>1.2.1-RELEASE</version>
</dependency>

编写配置：

canal:
  destination: heima # canal的集群名字，要与安装canal时设置的名称一致
  server: 192.168.150.101:11111 # canal服务地址

修改Item实体类

通过@Id、@Column、等注解完成Item与数据库表字段的映射：

package com.heima.item.pojo;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.Transient;

import javax.persistence.Column;
import java.util.Date;

@Data
@TableName("tb_item")
public class Item {
    @TableId(type = IdType.AUTO)
    @Id
    private Long id;//商品id
    @Column(name = "name")
    private String name;//商品名称
    private String title;//商品标题
    private Long price;//价格（分）
    private String image;//商品图片
    private String category;//分类名称
    private String brand;//品牌名称
    private String spec;//规格
    private Integer status;//商品状态 1-正常，2-下架
    private Date createTime;//创建时间
    private Date updateTime;//更新时间
    @TableField(exist = false)
    @Transient
    private Integer stock;
    @TableField(exist = false)
    @Transient
    private Integer sold;
}

@Transient：

可以将该字段不持久化至数据库表中

@TableField(exist = false)：

可以让Mybatis-plus忽略该字段，不将其作为SQL语句中的字段。

编写监听器

通过实现EntryHandler<T>接口编写监听器，监听Canal消息。注意两点：

实现类通过@CanalTable("tb_item")指定监听的表信息
EntryHandler的泛型是与表对应的实体类

package com.heima.item.canal;

import com.github.benmanes.caffeine.cache.Cache;
import com.heima.item.config.RedisHandler;
import com.heima.item.pojo.Item;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;
//指定要监听的表
@CanalTable("tb_item")
@Component
//Item是指定的被监听的实体类
public class ItemHandler implements EntryHandler<Item> {

    @Autowired
    private RedisHandler redisHandler;
    @Autowired
    private Cache<Long, Item> itemCache;

    @Override
    public void insert(Item item) {
        // 写数据到JVM进程缓存
        itemCache.put(item.getId(), item);
        // 写数据到redis
        redisHandler.saveItem(item);
    }

    @Override
    public void update(Item before, Item after) {
        // 写数据到JVM进程缓存
        itemCache.put(after.getId(), after);
        // 写数据到redis
        redisHandler.saveItem(after);
    }

    @Override
    public void delete(Item item) {
        // 删除数据到JVM进程缓存
        itemCache.invalidate(item.getId());
        // 删除数据到redis
        redisHandler.deleteItemById(item.getId());
    }
}

在这里对Redis的操作都封装到了RedisHandler这个对象中，是我们之前做缓存预热时编写的一个类，内容如下：

package com.heima.item.config;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.heima.item.pojo.Item;
import com.heima.item.pojo.ItemStock;
import com.heima.item.service.IItemService;
import com.heima.item.service.IItemStockService;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class RedisHandler implements InitializingBean {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private IItemService itemService;
    @Autowired
    private IItemStockService stockService;

    private static final ObjectMapper MAPPER = new ObjectMapper();

    @Override
    public void afterPropertiesSet() throws Exception {
        // 初始化缓存
        // 1.查询商品信息
        List<Item> itemList = itemService.list();
        // 2.放入缓存
        for (Item item : itemList) {
            // 2.1.item序列化为JSON
            String json = MAPPER.writeValueAsString(item);
            // 2.2.存入redis
            redisTemplate.opsForValue().set("item:id:" + item.getId(), json);
        }

        // 3.查询商品库存信息
        List<ItemStock> stockList = stockService.list();
        // 4.放入缓存
        for (ItemStock stock : stockList) {
            // 2.1.item序列化为JSON
            String json = MAPPER.writeValueAsString(stock);
            // 2.2.存入redis
            redisTemplate.opsForValue().set("item:stock:id:" + stock.getId(), json);
        }
    }

    public void saveItem(Item item) {
        try {
            String json = MAPPER.writeValueAsString(item);
            redisTemplate.opsForValue().set("item:id:" + item.getId(), json);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    public void deleteItemById(Long id) {
        redisTemplate.delete("item:id:" + id);
    }
}

Redis最佳实践

BigKey

Redis的Key虽然可以自定义，但最好遵循下面的几个最佳实践约定：

遵循基本格式：[业务名称]:[数据名]:[id]
长度不超过44字节
不包含特殊字符

例如：我们的登录业务，保存用户信息，其key可以设计成如下格式：

local : user :10

这样设计的好处：

可读性强
避免key冲突
方便管理
更节省内存： key是string类型，底层编码包含int、embstr和raw三种。embstr在小于44字节使用，采用连续内存空间，内存占用更小。当字节数大于44字节时，会转为raw模式存储，在raw模式下，内存空间不是连续的，而是采用一个指针指向了另外一段内存空间，在这段空间里存储SDS内容，这样空间不连续，访问的时候性能也就会收到影响，还有可能产生内存碎片

BigKey通常以Key的大小和Key中成员的数量来综合判定，例如：

Key本身的数据量过大：一个String类型的Key，它的值为5 MB
Key中的成员数过多：一个ZSET类型的Key，它的成员数量为10,000个
Key中成员的数据量过大：一个Hash类型的Key，它的成员数量虽然只有1,000个但这些成员的Value（值）总大小为100 MB

那么如何判断元素的大小呢？redis也给我们提供了命令

memory usage name 不过这种太耗费CPU了

所以我们用长度或列表大小来估算 strlen name llen list2

推荐值：

单个key的value小于10KB
对于集合类型的key，建议元素数量小于1000

BigKey的危害

网络阻塞
- 对BigKey执行读请求时，少量的QPS就可能导致带宽使用率被占满，导致Redis实例，乃至所在物理机变慢
数据倾斜
- BigKey所在的Redis实例内存使用率远超其他实例，无法使数据分片的内存资源达到均衡
Redis阻塞
- 对元素较多的hash、list、zset等做运算会耗时较旧，使主线程被阻塞
CPU压力
- 对BigKey的数据序列化和反序列化会导致CPU的使用率飙升，影响Redis实例和本机其它应用

如何发现BigKey

①redis-cli –bigkeys

利用redis-cli提供的–bigkeys参数，可以遍历分析所有key，并返回Key的整体统计信息与每个数据的Top1的big key

命令：redis-cli -a 密码 --bigkeys

②scan扫描

自己编程，利用scan扫描Redis中的所有key，利用strlen、hlen等命令判断key的长度（此处不建议使用MEMORY USAGE）

③第三方工具

利用第三方工具，如 Redis-Rdb-Tools 分析RDB快照文件，全面分析内存使用情况
https://github.com/sripathikrishnan/redis-rdb-tools

④网络监控

自定义工具，监控进出Redis的网络数据，超出预警值时主动告警
一般阿里云搭建的云服务器就有相关监控页面

如何删除BigKey

BigKey内存占用较多，即便时删除这样的key也需要耗费很长时间，导致Redis主线程阻塞，引发一系列问题。

redis 3.0 及以下版本

如果是集合类型，则遍历BigKey的元素，先逐个删除子元素，最后删除BigKey

Redis 4.0以后

Redis在4.0后提供了异步删除的命令：unlink

恰当的数据类型

例1：比如存储一个User对象，我们有三种存储方式：

①方式一：json字符串

user:1	{“name”: “Jack”, “age”: 21}

优点：实现简单粗暴

缺点：数据耦合，不够灵活

②方式二：字段打散

user:1:name	Jack
user:1:age	21

优点：可以灵活访问对象任意字段

缺点：占用空间大、没办法做统一控制

③方式三：hash（推荐）

user:1	name	jack
user:1	age	21

优点：底层使用ziplist，空间占用小，可以灵活访问对象的任意字段

缺点：代码相对复杂

例2：假如有hash类型的key，其中有100万对field和value，field是自增id，这个key存在什么问题？如何优化？

key	field	value
someKey	id:0	value0
	.....	.....
	id:999999	value999999

存在的问题：

hash的entry数量超过500时，会使用哈希表而不是ZipList，内存占用较多
- 62.23M
可以通过hash-max-ziplist-entries配置entry上限。但是如果entry过多就会导致BigKey问题

方案一–77.54M

拆分为string类型

key	value
id:0	value0
.....	.....
id:999999	value999999

存在的问题：

string结构底层没有太多内存优化，内存占用较多
想要批量获取这些数据比较麻烦

方案二-24.46M

拆分为小的hash，将 id / 100 作为key，将id % 100 作为field，这样每100个元素为一个Hash

key	field	value
key:0	id:00	value0
	.....	.....
	id:99	value99
key:1	id:00	value100
	.....	.....
	id:99	value199
....
key:9999	id:00	value999900
	.....	.....
	id:99	value999999

### 批处理优化

Redis提供了一些原生的批处理指令，不过每条指令只能处理特定的数据结构，例如mset，hmset

如果有对复杂数据类型的批处理需要，建议使用Pipeline

@Test
void testPipeline() {
    // 创建管道
    Pipeline pipeline = jedis.pipelined();
    long b = System.currentTimeMillis();
    for (int i = 1; i <= 100000; i++) {
        // 放入命令到管道
        pipeline.set("test:key_" + i, "value_" + i);
        if (i % 1000 == 0) {
            // 每放入1000条命令，批量执行
            pipeline.sync();
        }
    }
    long e = System.currentTimeMillis();
    System.out.println("time: " + (e - b));
}

针对于集群下的批处理

Spring集群环境下批处理代码，实现的是性能最好的并行slot

   @Test
    void testMSetInCluster() {
        Map<String, String> map = new HashMap<>(3);
        map.put("name", "Rose");
        map.put("age", "21");
        map.put("sex", "Female");
        stringRedisTemplate.opsForValue().multiSet(map);


        List<String> strings = stringRedisTemplate.opsForValue().multiGet(Arrays.asList("name", "age", "sex"));
        strings.forEach(System.out::println);

    }

Redis的持久化配置

Redis的持久化虽然可以保证数据安全，但也会带来很多额外的开销，因此持久化请遵循下列建议：

用来做缓存的Redis实例尽量不要开启持久化功能
建议关闭RDB持久化功能，使用AOF持久化
利用脚本定期在slave节点做RDB，实现数据备份
设置合理的rewrite阈值，避免频繁的bgrewrite
配置no-appendfsync-on-rewrite = yes，禁止在rewrite期间做aof，避免因AOF引起的阻塞
部署有关建议：
- Redis实例的物理机要预留足够内存，应对fork和rewrite
- 单个Redis实例内存上限不要太大，例如4G或8G。可以加快fork的速度、减少主从同步、数据迁移压力
- 不要与CPU密集型应用部署在一起
- 不要与高硬盘负载应用一起部署。例如：数据库、消息队列

Redis的慢查询

慢查询的阈值可以通过配置指定：

slowlog-log-slower-than：慢查询阈值，单位是微秒。默认是10000，建议1000

慢查询会被放入慢查询日志中，日志的长度有上限，可以通过配置指定：

slowlog-max-len：慢查询日志（本质是一个队列）的长度。默认是128，建议1000

知道了以上内容之后，那么咱们如何去查看慢查询日志列表呢：

slowlog len：查询慢查询日志长度
slowlog get [n]：读取n条慢查询日志
slowlog reset：清空慢查询列表

一些安全问题

Redis一定要设置密码
禁止线上使用下面命令：keys、flushall、flushdb、config set等命令。可以利用rename-command禁用。
bind：限制网卡，禁止外网网卡访问
开启防火墙
不要使用Root账户启动Redis
尽量不是有默认的端口

一些内存问题，还有集群主从

爱敲代码の鱼儿

http://19shuidiph.github.io/2023/10/13/redis-gao-ji-pian-hei-ma/

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源爱敲代码の鱼儿 !

无标签

JavaGuide自学目录4-IO&新特性

2023-10-17 JavaGuide

JavaGuide自学记录3--并发编程

2023-10-10 JavaGuide

Redis高级篇-黑马

Redis高级篇-黑马

分布式缓存

持久化

主从

哨兵（Sentinel）

分片集群

多级缓存

数据同步-canal

引入依赖：

编写配置：

修改Item实体类

编写监听器

Redis最佳实践

BigKey

BigKey的危害

如何发现BigKey

①redis-cli –bigkeys

②scan扫描

③第三方工具

④网络监控

如何删除BigKey

恰当的数据类型

例1：比如存储一个User对象，我们有三种存储方式：

①方式一：json字符串

②方式二：字段打散

③方式三：hash（推荐）

例2：假如有hash类型的key，其中有100万对field和value，field是自增id，这个key存在什么问题？如何优化？

方案一–77.54M

方案二-24.46M

Redis的持久化配置

Redis的慢查询

一些安全问题

一些内存问题，还有集群主从

你的赏识是我前进的动力