万古成空

2021-04-18 12:08 已编辑华东交通大学 Java

关注

差一点就深入理解redis

redis

简述

Redis 是一个开源（BSD许可）的，内存中的数据结构存储系统，它可以用作数据库、缓存和消息中

间件。它支持多种类型的数据结构，如字符串（strings），散列（hashes），列表（lists），

集合（sets），有序集合（sorted sets）与范围查询， bitmaps， hyperloglogs 和地理空间

（geospatial）索引半径查询。 Redis 内置了复制（replication），LUA脚本（Lua scripting），LRU驱动

事件（LRU eviction），事务（transactions）和不同级别的 磁盘持久化（persistence），并通过Redis

哨兵（Sentinel） 和自动分区（Cluster）提供高可用性（high availability）。

2008年，意大利一家创业公司Merzia的创始人Salvatore Sanfilippo为了避免MySQL的低性能，亲自定做

一个数据库，并于2009年开发完成，这个就是Redis。

从2010年3月15日起，Redis的开发工作由VMware主持。从2013年5月开始，Redis的开发由Pivotal赞助。

说明：Pivotal公司是由EMC和VMware联合成立的一家新公司**。**Pivotal希望为新一代的应用提供一个原

生的基础，建立在具有领导力的云和网络公司不断转型的IT特性之上。Pivotal的使命是推行这些创新，

提供给企业IT架构师和独立软件提供商。

redis支持：c,c++,java,python,go,c#等多种语言。

性能

下面是官方的bench-mark数据：

测试完成了50个并发执行100000个请求。

设置和获取的值是一个256字节字符串。

结果:读的速度是110000次/s,写的速度是81000次/s

Redis 单线程但性能依旧很快的主要原因有以下几点：

基于内存操作：Redis 的所有数据都存在内存中，因此所有的运算都是内存级别的，所以他的性能比较高；
数据结构简单：Redis 的数据结构比较简单，是为 Redis 专门设计的，而这些简单的数据结构的查找和操作的时间复杂度都是 O(1)，因此性能比较高；
多路复用和非阻塞 I/O：Redis 使用 I/O 多路复用功能来监听多个 socket 连接客户端，这样就可以使用一个线程连接来处理多个请求，减少线程切换带来的开销，同时也避免了 I/O 阻塞操作，从而大大提高了 Redis 的性能；
避免上下文切换：因为是单线程模型，因此就避免了不必要的上下文切换和多线程竞争，这就省去了多线程切换带来的时间和性能上的消耗，而且单线程不会导致死锁问题的发生。

数据类型与数据结构

string 、 hash 、 list 、 set 、 sorted set

这5种是暴露在使用者眼中，redis底层有下面5种数据结构：

dict
sds
ziplist
quicklist
skiplis

dict

    dict是一个用于维护key和value映射关系的数据结构，与很多语言中的Map或dictionary类似。Redis的一个database中所有key到value的映射，就是使用一个dict来维护的。不过，这只是它在Redis中的一个用途而已，它在Redis中被使用的地方还有很多。比如，一个Redis hash结构，当它的field较多时，便会采用dict来存储。再比如，Redis配合使用dict和skiplist来共同维护一个sorted set。这些细节我们后面再讨论，在本文中，我们集中精力讨论dict本身的实现。
    
typedef struct dictEntry {
    void *key;
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    struct dictEntry *next;
} dictEntry;

typedef struct dictType {
    unsigned int (*hashFunction)(const void *key);
    void *(*keyDup)(void *privdata, const void *key);
    void *(*valDup)(void *privdata, const void *obj);
    int (*keyCompare)(void *privdata, const void *key1, const void *key2);
    void (*keyDestructor)(void *privdata, void *key);
    void (*valDestructor)(void *privdata, void *obj);
} dictType;

/* This is our hash table structure. Every dictionary has two of this as we
 * implement incremental rehashing, for the old to the new table. */
typedef struct dictht {
    dictEntry **table;
    unsigned long size;
    unsigned long sizemask;
    unsigned long used;
} dictht;

typedef struct dict {
    dictType *type;
    void *privdata;
    dictht ht[2];
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    int iterators; /* number of iterators currently running */
} dict;


一个指向dictType结构的指针（type）。它通过自定义的方式使得dict的key和value能够存储任何类型的数据。

一个私有数据指针（privdata）。由调用者在创建dict的时候传进来。

两个哈希表（ht[2]）。只有在重哈希的过程中，ht[0]和ht[1]才都有效。而在平常情况下，只有ht[0]有效，ht[1]里面没有任何数据。hash时，ht[1]存放hash前的值，依次重hash放入ht[0],rehshidx+1就是ht[0]中已重hah个数。

当前重哈希索引（rehashidx）。如果rehashidx = -1，表示当前没有在重哈希过程中；否则，表示当前正在进行重哈希，且它的值记录了当前重哈希进行到哪一步了。
当前正在进行遍历的iterator的个数。这不是我们现在讨论的重点，暂时忽略。

sds
sds正是在Redis中被广泛使用的字符串结构，它的全称是Simple Dynamic String。与其它语言环境中出现的字符串相比，它具有如下显著的特点：
- 可动态扩展内存。sds表示的字符串其内容可以修改，也可以追加。在很多语言中字符串会分为mutable和immutable两种，显然sds属于mutable类型的。
- 二进制安全（Binary Safe）。sds能存储任意二进制数据，而不仅仅是可打印字符。
- 与传统的C语言字符串类型兼容。
在C语言中，字符串是以’\0’字符结尾（NULL结束符）的字符数组来存储的，通常表达为字符指针的形式（char *）。它不允许字节0出现在字符串中间，因此，它不能用来存储任意的二进制数据。

*sds=head+sds

head:
- len: 表示字符串的真正长度（不包含NULL结束符在内）。
- alloc: 表示字符串的最大容量（不包含最后多余的那个字节）。
- flags: 总是占用一个字节。其中的最低3个bit用来表示header的类型。header的类型共有5种，在sds.h中有常量定义。
*sds:存放值的char数组
ziplist

ziplist是list键、hash键以及zset键的底层实现之一（3.0之后list键已经不直接用ziplist和linkedlist作为底层实现了，取而代之的是quicklist）.

数据的粒度变大了，减少了内存的碎片数量,避免了不必要的内存浪费，每个元素动态分配空间。

quicklist

typedef struct quicklist {
    //指向头部(最左边)quicklist节点的指针
    quicklistNode *head;

    //指向尾部(最右边)quicklist节点的指针
    quicklistNode *tail;

    //ziplist中的entry节点计数器
    unsigned long count;        /* total count of all entries in all ziplists */

    //quicklist的quicklistNode节点计数器
    unsigned int len;           /* number of quicklistNodes */

    //保存ziplist的大小，配置文件设定，占16bits
    int fill : 16;              /* fill factor for individual nodes */

    //保存压缩程度值，配置文件设定，占16bits，0表示不压缩
    unsigned int compress : 16; /* depth of end nodes not to compress;0=off */
} quicklist;

typedef struct quicklistNode {
    struct quicklistNode *prev;     //前驱节点指针
    struct quicklistNode *next;     //后继节点指针

    //不设置压缩数据参数recompress时指向一个ziplist结构
    //设置压缩数据参数recompress指向quicklistLZF结构
    unsigned char *zl;

    //压缩列表ziplist的总长度
    unsigned int sz;                  /* ziplist size in bytes */

    //ziplist中包的节点数，占16 bits长度
    unsigned int count : 16;          /* count of items in ziplist */

    //表示是否采用了LZF压缩算法压缩quicklist节点，1表示压缩过，2表示没压缩，占2 bits长度
    unsigned int encoding : 2;        /* RAW==1 or LZF==2 */

    //表示一个quicklistNode节点是否采用ziplist结构保存数据，2表示压缩了，1表示没压缩，默认是2，占2bits长度
    unsigned int container : 2;       /* NONE==1 or ZIPLIST==2 */

    //标记quicklist节点的ziplist之前是否被解压缩过，占1bit长度
    //如果recompress为1，则等待被再次压缩
    unsigned int recompress : 1; /* was this node previous compressed? */

    //测试时使用
    unsigned int attempted_compress : 1; /* node can't compress; too small */

    //额外扩展位，占10bits长度
    unsigned int extra : 10; /* more bits to steal for future usage */
} quicklistNode;

使用双向链表，链表元素为ziplist。

skiplis

由多层组成，最底层为第1层，次底层为第2层，以此类推。层数不会超过一个固定的最大值Lmax。

每层都是一个有头节点的有序链表，第1层的链表包含跳表中的所有元素。

如果某个元素在第k层出现，那么在第1~k-1层也必定都会出现，但会按一定的概率p在第k+1层出现。

查找顺序由上到下：

找5：大于1，大于3，找到5

安装（centos）

下载：http://redis.io/
使用ftp工具拷贝到centos上，或使用rz命令（装yum install lrzsz）
解压：tar zxvf redis-6.0.9.tar.gz

编译：

yum -y install gcc-c++ autoconf automake
//升级gcc到9以上 
yum -y install centos-release-scl 
yum -y install devtoolset-9-gcc devtoolset-9-gcc-c++ devtoolset-9-binutils 
//临时将此时的gcc版本改为9 
scl enable devtoolset-9 bash 
//或永久改变 
echo "source /opt/rh/devtoolset-9/enable" >>/etc/profile source /etc/profile
//检验更新到9
gcc -v
//进入解压后redis，编译
cd redis-6.0.9
make

安装:

//创建安装目录
mkdir -p /usr/local/redis
//指定安装目录安装
make PREFIX=/usr/local/redis/ install

运行

//Redis-cli :客户端
//Redis-server :服务器端
//进入安装目录
cd /usr/local/redis/bin

//启动（无密码） 后面可以接  .cnf配置文件
./redis-service

//连接客户端
./redis-cli 

//退出客户端
exit或ctrl+c

//关闭redis服务
ps -ef|grep redis
kill -9 reids进程号

docker安装redis

找到镜像：https://hub.daocloud.io/repos/beb958f9-ffb6-4f68-817b-c17e1ff476c3

centos中：

docker pull daocloud.io/library/redis:3.2.9
//查看镜像id
docker images

//第一次运行 映射主机端口6380，因为我本机装了redis，所以主机的6379端口不能给docker
//docker run -itd --name redis -p 6380:6379  redis镜像id
docker run -itd --name redis-test -p 6378:6379 34

//连接redis客户端
docker exec -it redis /bin/bash

//查看容器id,加-a看到停止的容器，不加只能看到运行中容器
docker ps -a

//关闭
docker stop 容器id

//下次要启动
docker start 容器id

这里我将本机redis.conf文件通过：docker cp 容器id:/usr/local/bin

运行centos

进入安装路径：/usr/local/bin

使用命令：./redis-server

连接：./redis-cli

配置redis.conf

取消IP绑定：#bind 127.0.0.1
设置后台启动：daemonize yes
关闭保护模式：protected-mode no
rdb持久化：dbfilename dump.rdb
密码：requirepass root

连接：

自带客户端：redis-cli:
图像化界面

多线程

io-threads-do-reads ：多线程默认是禁用的，只是用主线程，需要额外开启。

io-threads ：开启多线程后，需要设置线程数，否则不生效。关于线程数的设置，官方有一个建议：4核的机器建议设置为2或3个线程，8核的建议设置为6个线程，线程数一定要小于机器核数。还需要注意的是，线程数并不是越大越好，官方认为超过了8个基本就没什么意义了。

关于单线程：

redis 4：当Redis需要删除一个很大的数据时（比如结构很复杂的hash），因为是单线程同步操作，这就会导致 Redis 服务卡顿，于是在 Redis 4.0 中就新增了多线程的模块，当然此版本中的多线程主要是为了解决删除数据效率比较低的问题的，他的相关指令有以下三个：

unlink key
flushdb async
flushall async

执行示例如下所示：

> unlink key # 后台删除某个 key
> OK # 执行成功
> flushall async # 清空所有数据
> OK # 执行成功

redis虽然从6.x版本开始支持多线程，但是本质上还是单线程，redis采用io多路复用，多线程io读写数据，但是命令的执行还是单线程（主线程负责），因此redis还是线程安全。

关系型数据库

采用关系模型来组织数据的数据库，关系模型就是二维表格模型。一张二维表的表名就是关系，二维表中的一行就是一条记录，二维表中的一列就是一个字段。

优点

容易理解
使用方便，通用的sql语言
易于维护，丰富的完整性(实体完整性、参照完整性和用户定义的完整性)大大降低了数据冗余和数据不一致的概率

缺点

磁盘I/O是并发的瓶颈
海量数据查询效率低
横向扩展困难，无法简单的通过添加硬件和服务节点来扩展性能和负载能力，当需要对数据库进行升级和扩展时，需要停机维护和数据迁移
多表的关联查询以及复杂的数据分析类型的复杂sql查询，性能欠佳。因为要保证acid（原子性，一致性，隔离性，持久性），必须按照三范式设计。

数据库 Orcale，Sql Server，MySql，DB2

非关系型数据库

非关系型，分布式，一般不保证遵循ACID原则的数据存储系统。键值对存储，结构不固定。

优点

根据需要添加字段，不需要多表联查。仅需id取出对应的value
适用于SNS（社会化网络服务软件。比如facebook，微博）
严格上讲不是一种数据库，而是一种数据结构化存储方法的集合

缺点

只适合存储一些较为简单的数据
不合适复杂查询的数据
不合适持久存储海量数据

代表

K-V：Redis（类型多），Memcache （类型少，大数据量性能高）
文档：MongoDB
搜索：Elasticsearch，Solr
可扩展性分布式：HBase

redis-cli操作指令

string字符串

set :添加一条String类型数据

get :获取一条String类型数据

mset :添加多条String类型数据

mget :获取多条String类型数据

del:删除key，通用

list列表

lpush :左添加(头)list类型数据

rpush :右添加(尾)类型数据

lrange : 获取list类型数据start起始下标 end结束下标包含关系，长度超过列表长从起始到后面全部打印

llen :获取条数

lrem :删除列表中几个指定list类型数据

hash哈希

hset :添加一条hash类型数据

hget :获取一条hash类型数据

hmset :添加多条hash类型数据

hmget :获取多条hash类型数据

hgetAll :获取指定所有hash类型数据

hdel :删除指定hash类型数据(一条或多条)

set集合

sadd :添加set类型数据

smembers :获取set类型数据

scard :获取条数

srem :删除数据

zset有序集合

sorted set是通过分数值来进行排序的，分数值越大，越靠后。

zadd :添加sorted set类型数据

zrange :获取sorted set类型数据

zcard :获取条数

zrem :删除数据

zadd需要将Float或者Double类型分数值参数，放置在值参数之前

zadd score 数字1 值1 数字2 值2 数字3 值3

失效时间

ex：几秒后过期

px：几毫秒后过期

127.0.0.1:6379> set ppl shadiao ex 2
OK
127.0.0.1:6379> get ppl
(nil)
127.0.0.1:6379>

两秒后失效取不到值

127.0.0.1:6379> set ppl ppDog px 5000
OK
127.0.0.1:6379> get ppl
"ppDog"
127.0.0.1:6379> get ppl
(nil)

5秒内有效

查看有效时间：

127.0.0.1:6379> set lwf good
OK
127.0.0.1:6379> set ppl dog ex 20
OK
127.0.0.1:6379> ttl lwf
(integer) -1
127.0.0.1:6379> ttl ppl
(integer) 2
127.0.0.1:6379> ttl ppl
(integer) -2

默认不失效，-1

正数表示有效时间

-2表过期

spring boot使用redis

springboot2.2.X：用Lettuce替换Jedis操作Redis缓存

这两个都是用于提供连接Redis的客户端。

Jedis是直接连接Redis，非线程安全，在性能上，每个线程都去拿自己的 Jedis 实例，当连接数量增多时，资源消耗阶梯式增大，连接成本就较高了。

Lettuce的连接是基于Netty的，Netty 是一个多线程、事件驱动的 I/O 框架。连接实例可以在多个线程间共享，当多线程使用同一连接实例时，是线程安全的。

jedis

application.yml

spring:
  application:
    name: redis
  level: debug
  redis:
    host: 192.168.10.100
    pool:
      max-active: 8
      max-idle: 8
      max-wait: 1000
      min-idle: 3
    password: root
    database: 0
    port: 6379

pom.xml

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.0</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <java.version>11</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

jedis配置类

@Configuration
public class RedisPool {
    @Value("${spring.redis.host}")
    private String host;
    @Value("${spring.redis.port}")
    private Integer port;
    @Value("${spring.redis.password}")
    private String password;
    @Value("${spring.redis.pool.max-active}")
    private Integer maxAct;
    @Value("${spring.redis.pool.max-idle}")
    private Integer maxIdle;
    @Value("${spring.redis.pool.max-wait}")
    private Integer maxWait;
    @Value("${spring.redis.pool.min-idle}")
    private Integer minIdle;
    @Bean
    public JedisPool getPool(){
        JedisPoolConfig config=new JedisPoolConfig();
        config.setMaxIdle(maxIdle);
        config.setMaxWaitMillis(maxWait);
        config.setMinIdle(minIdle);
        config.setMaxTotal(maxAct);
        JedisPool pool = new JedisPool(config,host,port,maxWait,password, 0, false);
        return pool;
    }
}

测试

@SpringBootTest
@RunWith(SpringRunner.class)
public class AppTest 
{
    /**
     * Rigorous Test :-)
     */
    @Resource
    private JedisPool jedisPool;
    private Jedis jedis=null;
    @Before
    public void before(){
        jedis=jedisPool.getResource();
    }

    /**
     * string操作
     */
    @Test
    public void string() {
        System.out.println(jedis.ping());
        //存string,返回ok或0
        System.out.println(jedis.set("ppl", "皮皮狗"));
        System.out.println(jedis.setnx("ppl", "狗子"));
        //存值，ex（30）设置30秒失效，nx（）key不存在才能添加
        System.out.println(jedis.set("lsh", "shadiao", SetParams.setParams().ex(2).nx()));
        //按key取值
        System.out.println(jedis.get("lsh"));
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //获取当前数据库
        System.out.println(jedis.getDB());
        //获取所有key，key用正则匹配
        System.out.println(jedis.keys("*"));
    }

    /**
     * 计时器设置
     * @throws InterruptedException
     */
    @Test
    public void testTimer() throws InterruptedException {
        //2秒过期
        jedis.expire("ppl", 2);
        Thread.sleep(2001);
        System.out.println("ppl存活:"+jedis.exists("ppl"));
        //设置毫秒后过期，在cli中-1表不过期，这里设置不能负数，不适用ex，px就是不过期
        jedis.set("ppl", "皮皮狗");
        jedis.pexpire("ppl", 2);
        //生存时间
        System.out.println(jedis.ttl("ppl"));
        System.out.println("ppl"+jedis.get("ppl")+"存活:"+jedis.exists("ppl"));
    }

    /**
     * hash操作
     */
    @Test
    public void hash(){
        Map<String,String> map=new HashMap<>();
        map.put("ppl", "pp dog");
        map.put("zyr", "日天");
        //加map
        jedis.hset("hdjd", map);
        //取值, <key,mapKey>
        System.out.println(jedis.hget("hdjd", "ppl"));
        jedis.hgetAll("hdjd").forEach((k,v)->{
            System.out.println(k+":"+v);
        });
        //删除一个hash的一个map
        jedis.hdel("hdjd", "ppl");
        jedis.hgetAll("hdjd").forEach((k,v)->{
            System.out.println(k+":"+v);
        });
        //全删除
        jedis.del("hdja");
        //加一个
        jedis.hset("lwf", "city" ,"上饶");
        System.out.println(jedis.hget("lwf", "city"));
        //插入，key不存在才能插入并返回1，否则返回0
        System.out.println(jedis.hsetnx("lwf", "height", "3 m 3 cm"));
        //key存在
        System.out.println(jedis.hsetnx("lwf", "city", "南昌"));
        //清空
        jedis.flushAll();
    }

    /**
     * 列表
     */
    @Test
    public void list(){
        //依次加入列表首,倒序排入
        jedis.lpush("index", "lwf","ppl","zyr","lsh","xyh");
        //一次加入表尾,顺序插入尾
        jedis.rpush("last", "工具人","xyh","扑街");
        //切片,开始，结束，两边包含，stop大于最大下标，则全部输出；负数则是从倒数第几结束
        System.out.println("结束下标小于最大下标");
        jedis.lrange("index", 0, 4).forEach(System.out::println);
        System.out.println("结束下标小于0，显示到倒数第二位");
        jedis.lrange("index", 0, -2).forEach(System.out::println);
        //结束下标负数，且指向真实下标小于开始
        System.out.println("结束表在开始表左,没有值");
        jedis.lrange("index", 0, -6).forEach(System.out::println);
        //表长
        System.out.println("表长");
        System.out.println(jedis.llen("last"));
        System.out.println("删除指定值几个，count多就全删此key值");
        jedis.lrem("index", 2,"ppl");
        System.out.println("全能删除,删除返回1，否则0");
        System.out.println("删除了:"+jedis.del("index"));
        System.out.println(jedis.del("last"));
    }

    /**
     * 集合
     */
    @Test
    public void set(){
        //无序不重复
        jedis.sadd("key", "lll","ppp","lll","fff");
        System.out.println(jedis.scard("key"));
        jedis.smembers("key").forEach(System.out::println);
        //删除
        jedis.srem("key", "fff");
        //删掉key
        jedis.del("key");
    }
    @Test
    public void zSet(){
        //score属性为排序依据（值大小排)
        jedis.zadd("jx", 1D, "lwf");
        jedis.zadd("jx", 9D, "lsh");
        jedis.zadd("jx", 6D, "zyr");
        jedis.zadd("jx", 4D, "plp");
        System.out.println("条数"+jedis.zcard("jx"));
        //和列表操作相同
        jedis.zrange("jx", 0, -3).forEach(System.out::println);
        //删除,返回删除个数
        System.out.println(jedis.zrem("jx", "plp","lsh"));

    }
    @After
    public void after(){
        jedis.close();
        jedisPool.close();
    }
}

lettuce

application.yml

spring:
  redis:
    host: 192.168.10.102
    port: 6379
    database: 0
    timeout: 10000ms
    password: root
    lettuce:
      pool:
        max-active: 1024
        max-wait: 10000ms
        max-idle: 200
        min-idle: 5
    sentinel:
      master: mymaster
      nodes: 192.168.10.100:26379,192.168.10.101:26379

pom.xml

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.0</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    
    <properties>
        <java.version>11</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <!-- commons-pool2 对象池依赖 -->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

RedisTemplate配置类

@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate<String,Object> getBean(LettuceConnectionFactory factory){
        RedisTemplate<String, Object> template = new RedisTemplate();
        template.setConnectionFactory(factory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
        return template;
    }
}

测试

@SpringBootTest
class RedisTemplateApplicationTests {
    @Autowired
    private RedisTemplate<String,Object> redisTemplate;
    @Test
    public void contextLoads() {
        ValueOperations<String, Object> ops = redisTemplate.opsForValue();
        ops.set("ppl", "沙雕", 12, TimeUnit.SECONDS);
        System.out.println(ops.setIfAbsent("ppl", "，皮皮狗"));
        System.out.println("ops.get(\"ppl\") = " + ops.get("ppl"));
    }
}

redis持久化

rdb

RDB其实就是把数据以快照的形式保存在磁盘上。

什么是快照呢，你可以理解成把当前时刻的数据拍成一张照片保存下来。 RDB持久化是指在指定的时间间隔内将内存中的数据集快照写入磁盘。也是默认的持久化方式，这种方式是就是将内存中数据以快照的方式写入到二进制文件中,默认的文件名为dump.rdb。

既然RDB机制是通过把某个时刻的所有数据生成一个快照来保存，那么就应该有一种触发机制，是实现这个过程。对于RDB来说，提供了三种机制：save、bgsave、自动化。

save： 阻塞式保存，使用该命令将数据写入dump.rdb。执行save命令期间，Redis不能处理其他命令，直到 RDB过程完成为止。

bgsave 后台保存

自动化 save 900 1 ：表示900秒内检测到1个key发生变化(写入操作即增删改)，就会自动进行bgsave的命令存储快照，以次类推下面两个命令的含义。

优点：

RDB文件紧凑，全量备份，非常适合用于进行备份和灾难恢复。
生成RDB文件的时候，redis主进程会fork()一个子进程来处理所有保存工作，主进程不需要进行任何磁盘IO操作。
RDB 在恢复大数据集时的速度比 AOF 的恢复速度要快。

缺点：

RDB快照是一次全量备份，存储的是内存数据的二进制序列化形式，存储上非常紧凑。当进行快照持久化时，会开启一个子进程专门负责快照持久化，子进程会拥有父进程的内存数据，父进程修改内存子进程不会反应出来，所以在快照持久化期间修改的数据不会被保存，可能丢失数据。

aof

appendonly ：开启AOF，默认关闭
appendfilename ：AOF存储的文件名称

优点：

AOF可以更好的保护数据不丢失，一般AOF会每隔1秒，通过一个后台线程执行一次fsync操作，最多丢失1秒钟的数据。

AOF日志文件没有任何磁盘寻址的开销，写入性能非常高，文件不容易破损。 AOF日志文件的命令通过非常可读的方式进行记录，这个特性非常适合做灾难性的误删除的紧急恢复。

缺点：

对于同一份数据来说，AOF日志文件通常比RDB数据快照文件更大 AOF开启后，支持的写QPS会比RDB支持的写QPS低，因为AOF一般会配置成每秒fsync一次日志文件。

配合使用

rdb文件小，但是数据容易丢失。

aof数据丢失少，但是文件大。

当两者配合使用时；

rdb持久化存储数据，而aof存储两次写入rdb文件之间的操作。

RDB代表全量数据，AOF代表增量数据

单点故障解决

一台电脑redis，如果服务器宕机，就会引发单点故障，redis服务无法访问。

为了解决单点故障，这里有2种策略：主从，集群。

主从

redis一般被用来做缓存，存放热点数据，所以读远远大于写，所以只需要主节点可以读写操作，从节点复制备份主节点数据，并提供读取数据服务。

这里使用了3台redis服务器：由于半数不可用，3台服务器，两台挂掉才会3台不可用。而4台，两台挂掉，全部不可用；这里就节省了一台redis从节点服务器。

这里就需要哨兵。

当服务器宕机：

主节点：这里使用心跳（默认30秒ping没有回应认为主节点挂掉），哨兵选举出master，master开始从从节点中选出主节点，

选出主节点后，会再判断原主节点是否可通信，可通行则不更改主节点，否则更改主节点为哨兵master选举出的节点。并将原主节点，以及其他从节点全部挂到新主节点下。
从节点：哨兵发现宕机直接踢出去。

这里看出主节点为192.168.10.100也就是本机,有一个从节点

开启主从配置：redis.conf

主节点

注释掉：bind 127.0.0.1
关闭保护模式：protected-mode no
后台：daemonize yes
主节点密码：masterauth root

从节点

注释掉：bind 127.0.0.1
关闭保护模式：protected-mode no
后台：daemonize yes
主节点密码：masterauth root
主节点ip 端口
replicaof 192.168.10.100 6379

哨兵：sentinel.conf（拷贝到每个redis的安装目录下）

主节点ip 端口 需要几个哨兵认为它挂了才开始选举
sentinel monitor mymaster 192.168.10.100 6379 2
设置访问主节点密码，心跳通信ping pong
sentinel auth-pass mymaster root
关闭保护模式
protected-mode no
后台启动
daemonize yes

启动

主节点

./redis-server redis.conf
./redis-sentinel sentinel.conf

从节点

./redis-server redis.conf
./redis-sentinel sentinel.conf

集群

3个主节点形成集群，首先节点1和节点2握手，成功就加入集群，节点1再与节点3握手。

集群中，当一台主节点挂掉了，这台主节点的数据就会分配给可用的一台主节点。所以当节点数少时，数据转移压力就大，节点多数据转移量就小。

hash环指派

该节点宕机，数据会被迁移到该环中下一个节点；

（宕机节点编号+1）%节点数
槽指派(数据存槽里)

该节点宕机时，该节点的槽一分为二分别给相邻主节点

常见问题

缓存穿透

缓存穿透是指查询缓存和DB中都不存在的数据。比如通过id查询商品信息，id一般大于0，攻

击者会故意传id为-1去查询，由于缓存是不命中则从DB中获取数据，这将会导致每次缓存都不命中数据

导致每个请求都访问DB，造成缓存穿透。

解决方案：

利用互斥锁，缓存失效的时候，先去获得锁，得到锁了，再去请求数据库。没得到锁，则休眠一段

时间重试。
数据库数据数据没有查询到，返回null存redis并设置过期时间。
用户行为判断，多次访问不存在数据，就判断恶意攻击，一段时间不给用户访问数据库的操作，只能访问redis数据。
提供一个能迅速判断请求是否有效的拦截机制，比如，利用布隆过滤器，内部维护一系列合法有效

的key。迅速判断出，请求所携带的Key是否合法有效。如果不合法，则直接返回。