一、NOSQL

1、概述

泛指非关系型数据库

关系型数据库：表格、行、列

2、特点

方便拓展
大数据量，高性能（1s写8w次，读取11w，NOSQL的缓存记录级，是一种细粒度的缓存，性能比较高）
数据类型是多样型的（不需要事先设计数据库，随取随用）
传统的RDBMS和NOSQL的关系

传统的数据库

--结构化组织

--sql

--数据和关系都存在单独的表中

nosql

--不仅仅是数据

--没有固定的查询语言

--键值对存储、列存储、文档存储、图像数据库

--最终一致性

--CAP && BASE (异地多活)

--高可用、高可拓、高性能

3、四大分类

kv键值对
文档型数据库

　　　　MongoDB是基于分布式文件存储的数据库，处理大量的文档，是一个介于关系型和非关系型中间的产品

　　　　ConthDB

列存储数据库

　　　　HBase

　　　　分布式文件系统

图关系数据库

分类	Examples举例	典型应用场景	数据模型	优点	缺点
键值（key-value）	Tokyo Cabinet/Tyrant, Redis, Voldemort, Oracle BDB	内容缓存，主要用于处理大量数据的高访问负载，也用于一些日志系统等等。	Key 指向 Value 的键值对，通常用hash table来实现	查找速度快	数据无结构化，通常只被当作字符串或者二进制数据
列存储数据库	Cassandra, HBase, Riak	分布式的文件系统	以列簇式存储，将同一列数据存在一起	查找速度快，可扩展性强，更容易进行分布式扩展	功能相对局限
文档型数据库	CouchDB, MongoDb	Web应用（与Key-Value类似，Value是结构化的，不同的是数据库能够了解Value的内容）	Key-Value对应的键值对，Value为结构化数据	数据结构要求不严格，表结构可变，不需要像关系型数据库一样需要预先定义表结构	查询性能不高，而且缺乏统一的查询语法。
图形(Graph)数据库	Neo4J, InfoGrid, Infinite Graph	社交网络，推荐系统等。专注于构建关系图谱	图结构	利用图结构相关算法。比如最短路径寻址，N度关系查找等	很多时候需要对整个图做计算才能得出需要的信息，而且这种结构不太好做分布式的集群方案。

二、redis

1、redis能干吗

内存存储、持久化(rdb、aof)，内存中是断电即失

效率高，可以用于高速缓存

发布订阅系统

地图信息分析

计时器、计数器

......

2、redis安装

redis-benchmark测试性能

redis-benchmark -h localhost -p 6379 -c 100 -n 100000

序号	选项	描述	默认值
1	-h	指定服务器主机名	127.0.0.1
2	-p	指定服务器端口	6379
3	-s	指定服务器 socket
4	-c	指定并发连接数	50
5	-n	指定请求数	10000
6	-d	以字节的形式指定 SET/GET 值的数据大小	2
7	-k	1=keep alive 0=reconnect	1
8	-r	SET/GET/INCR 使用随机 key, SADD 使用随机值
9	-P	通过管道传输 <numreq> 请求	1
10	-q	强制退出 redis。仅显示 query/sec 值
11	--csv	以 CSV 格式输出
12	-l（L 的小写字母）	生成循环，永久执行测试
13	-t	仅运行以逗号分隔的测试命令列表。
14	-I（i 的大写字母）	Idle

10w请求在3.36s完成

100个并行的客户端

每次写入3比特

1台服务器处理请求，单机性能

100% 所有请求在36毫秒完成

切换数据库

redis-cli　　　　//连接客户端
select 3        //选择数据库
flushdb　　　　　//清空当前数据库
flushall　　　　 //清空所有数据库

redis是单线程的

redis是基于内存操作的，cpu不是性能redis的性能瓶颈，而是根据机器的内存和网络带宽决定的，既然可以使用单线程就使用单线程了

为什么单线程还这么快

redis是c语音写的，每秒qps是10w+，完全不比memecache差

redis将数据写在内存中，所以单线程更快，因为多线程（切换线程：上下文的切换），对于内存系统来说，没有上下文切换就是效率最高的，多次读写都是在一个cpu上

单点登录

三、五大数据类型

redis-key

expire name 10  　　　　 //设置过期时间  单点登录
ttl name 　　　　　　　　 //查看剩余时间
type name 　　　　　　　　//查看数据类型

1、string

append key1 11   　　　　　　　　   //拼接
strlen key1   　　　　　　　　　  　//长度
incr key1 　　　　　　　　　　　　   //自增
decr key1     　　 　　　　　　　　 //自减
incrby key1 10 　　　　　　　　　 　//每次加10
decrby key1 10　　 　　　　　　　　 //每次减10
getrange name 0 1 　　　　　　　　 //截取字符串
getrange name 0 -1 　　　　　　　　//全部字符串
setrange name 0 xx 　　　　　　　　//指定位置开始替换
setex key1 30 "hello" 　　　　    //30秒内赋值（时间段内 set不会被覆盖）
setnx key1　"redis"　  　　　　    //不存在再设置 返回-1
mset k1 k2 k3          　　　　   //同时设置多个值（原子性操作，同时失败或者同时成功）
mget k1 k2 k3 　　　　            //同时获取多个值
set use:1 [age:1,goal:40]  　　  //对象
mset user:1:name zhangsan user:1:age 2  //对象
getset db "redis"      　　　　   //先get再set

2、list

可以完成栈、队列阻塞队列

lpush list one     　　　 　　　　　　　//赋值
lrange 0 -1　　　　 　　　 　　　　　　　//获取全部
lrange 0 1 　　　　 　　　 　　　　　　　//倒叙获取
rpush list righr　　　　　　　　　　　　//从右边插入
lpop list 　　　　  　　　　　　　　　　 //移除列表第一个元素
rpop list　　　　  　　　　  　　　　　　//移除列表最后一个
lindex list 1　　　　　　　  　　　　　　//获取下标为1的元素
llen list　　　　　　　　　　 　　　　　　//长度
lrom list 1 three　　　　   　　　　　　//移除1个three 
ltrim list i 2      　　   　　　　　　//截取指定下标
rpoplpush list list1       　　　　　　//list右边出移出到list1
exists list　　　　　　      　　　　　　//是否存在
lset list 0 two            　　　　　　//下标重新赋值(不存在更新会报错)
linsert list before "one" "hello"    //在one之前插入

3、set

例如微博的共同关注等

sadd myset "hello"　　　　　　 //集合添加元素
smembers myset　　　　　　　　 //查看
sismember myset "hello"　　　//是否存在
srom myset hello　　　　　　  //移出hello
acard myset　　　　　　　　    //有多少元素、
srandmember myset 　　       //随机取数据
srandmember myset 1　　　　   //随机取1个数据
spop myset　　　　　　　　　　  //随机删除元素
smove myset myset2 "元素"    //转移元素
sdiff myset1 myset2         //差集 前者集合中后者没有的
sinter myset1 myset2　　　　  //交集 
sunion myset1 myset2 　　　　 //并集

4、hash

key-map集合

hset myhash field1 hello　　　　　　　　 　 //赋值
hget myhash field1　　　　　　　　　　　　 　//获取
hmset myhash field1 hello field2 world 　  //批量赋值
hmget myhash field1 field2　　　　　　　 　 //批量获取
hgetall myhash 　　　　　　　　　　　　　 　 //获取所有
hlen myhash　　　　　　　　　　　　　　　　　 //长度
hexists myhash　field1 　　　　　　　　     //指定是否存在
hkeys myshsh　　　　　　　　　　　　　　　　 //全部key
hvals myhash　　　　　　　　　　　　　　　　 //全部valus
hincrby myhash field1 1　　　　　　　　     //指定增1
jdecrby myhash field1 -1　　　　　　　　　  //减1

5、zset 有序集合

set ---》 key value

zset ---》 key sorted value

zadd myset1 1 one　　　　　　　　　　 //添加
zadd myset1 1 one 2 two 3 three　　//添加多数据
zrange myset　　　　　　　　　　　　　 //查看

zadd salary 2500 lee　　　　　　　　　
zadd salary 3000 zhang
zadd salary 2000 wang
zrangebyscore salary -inf +inf　　　 //从小到大排序
zrevrange salary 0 -1　　　　　　　　 //反排序
zrange salary 0 -1　　　　　　　　　　//查看
zrom salary lee　　　　　　　　　　　　//移除
zcard salary　　　　　　　　　　　　 　// 数量
zcount mysey 1 3　　　　　　　　　　　//指定区间的数量

四、特殊数据类型

1、geospitial地理位置

getadd key 纬度经度名称

getadd china:city 121.47.31.23 shanghai  //添加
geopos china:city shanghai 　　　　　　　　 //获取
geodist china:city shanghai beijing      //距离  m km ml盈利 ft英尺
georadius china:city 110 30 1000 km　　　　//经度 纬度 半径 单位
georadius china:ciyt 110 30 1100 km count 1  //取一个
georadius china:city 110 30 1100 km asc/desc  //由近到远 由远及近
georadiusbymember china:city beijing 1000 km //指定元素周围的其他元素

返回11个字符的字符串

他们可以缩短从右边的字符。它将失去精度，但仍将指向同一地区。

它可以在 geohash.org 网站使用，网址 http://geohash.org/<geohash-string>。

查询例子：Geohash - geohash.org/sqdtr74hyu0.

与类似的前缀字符串是附近，但相反的是不正确的，这是可能的，用不同的前缀字符串附近。

geohash china:city beijing       //返回的是经纬度字符串

geo底层实现是zset，使用zset的命令可以操作geo删除等

2、hyperloglog

基数：不重复的元素个数

可以接受误差

优点：

占用内存固定，2^64不同元素的基数，只需要12kb内存

PFadd mykey a b c d e f g　　　　　　//添加
PFcount mykey 　　　　　　　　　　　　 //数量
PFmerge mykey3 mykey1 mykey2　　　　//合并

3、bitmap

位存储，两个状态的都可以使用

setbit sign 1 0      //赋值
getbit sign 1　　　　　//获取
bigcount sign　　　　  //计数

五、事务

redis单条命令保持原子性，但是事务不保持原子性

redis事务没有隔离的概念，所有的事务并没有直接被执行，只有发起命令的时候才执行

一个事务中的命令都会被序列化，事务执行中会按照顺序执行

一次性，顺序性，排他性

开启事务
命令入队
执行事务

multi　　　　　//开启
set k1 v1 　　//入栈（命令入队）
exec　　　　　 //执行
discard　　　　//取消执行

编译时异常，直接报错

运行时异常，虽然一条失败，其他的可以成功

悲观锁：

无论做什么都加锁

乐观锁：

什么时候都不会出现问题，拿数据的时候不会加锁，更新数据的时候判断在此期间是否有人修改这个数据

获取version

更新的时候比较version

//一般情况 一个线程
set money 100
set out 0
watch money　　　　　　//监视对象
multi
decrby money 10
incrby out 10
exec

//多个线程
set money 100
set out 0
watch money　　　　
multi
decrby money 10
incrby out 10

set out 0
watch money　　　　　
multi
decrby money 10
incrby out 10

//再去执行第一个exec，显示null，执行失败

如果失败，解锁重新获取新的值即可

unwatch

六、Jedis

1、导入依赖

    <!-- jedis -->
    <dependencies>
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>4.3.1</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.75</version>
        </dependency>
    </dependencies>

2、测试

public class TestDemo1 {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        //学过的命令可直接使用
        System.out.println(jedis.ping());
    }
}

3、常用

import redis.clients.jedis.Jedis;

public class TestPassword {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1", 6379);

        //验证密码，如果没有设置密码这段代码省略
//        jedis.auth("password");

        jedis.connect(); //连接
        jedis.disconnect(); //断开连接

        jedis.flushAll(); //清空所有的key
    }
}

import redis.clients.jedis.Jedis;

import java.util.concurrent.TimeUnit;

public class TestString {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1", 6379);

        jedis.flushDB();
        System.out.println("===========增加数据===========");
        System.out.println(jedis.set("key1","value1"));
        System.out.println(jedis.set("key2","value2"));
        System.out.println(jedis.set("key3", "value3"));
        System.out.println("删除键key2:"+jedis.del("key2"));
        System.out.println("获取键key2:"+jedis.get("key2"));
        System.out.println("修改key1:"+jedis.set("key1", "value1Changed"));
        System.out.println("获取key1的值："+jedis.get("key1"));
        System.out.println("在key3后面加入值："+jedis.append("key3", "End"));
        System.out.println("key3的值："+jedis.get("key3"));
        System.out.println("增加多个键值对："+jedis.mset("key01","value01","key02","value02","key03","value03"));
        System.out.println("获取多个键值对："+jedis.mget("key01","key02","key03"));
        System.out.println("获取多个键值对："+jedis.mget("key01","key02","key03","key04"));
        System.out.println("删除多个键值对："+jedis.del("key01","key02"));
        System.out.println("获取多个键值对："+jedis.mget("key01","key02","key03"));

        jedis.flushDB();
        System.out.println("===========新增键值对防止覆盖原先值==============");
        System.out.println(jedis.setnx("key1", "value1"));
        System.out.println(jedis.setnx("key2", "value2"));
        System.out.println(jedis.setnx("key2", "value2-new"));
        System.out.println(jedis.get("key1"));
        System.out.println(jedis.get("key2"));

        System.out.println("===========新增键值对并设置有效时间=============");
        System.out.println(jedis.setex("key3", 2, "value3"));
        System.out.println(jedis.get("key3"));
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(jedis.get("key3"));

        System.out.println("===========获取原值，更新为新值==========");
        System.out.println(jedis.getSet("key2", "key2GetSet"));
        System.out.println(jedis.get("key2"));

        System.out.println("获得key2的值的字串："+jedis.getrange("key2", 2, 4));
    }
}

import redis.clients.jedis.Jedis;

public class TestSet {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        jedis.flushDB();
        System.out.println("============向集合中添加元素（不重复）============");
        System.out.println(jedis.sadd("eleSet", "e1","e2","e4","e3","e0","e8","e7","e5"));
        System.out.println(jedis.sadd("eleSet", "e6"));
        System.out.println(jedis.sadd("eleSet", "e6"));
        System.out.println("eleSet的所有元素为："+jedis.smembers("eleSet"));
        System.out.println("删除一个元素e0："+jedis.srem("eleSet", "e0"));
        System.out.println("eleSet的所有元素为："+jedis.smembers("eleSet"));
        System.out.println("删除两个元素e7和e6："+jedis.srem("eleSet", "e7","e6"));
        System.out.println("eleSet的所有元素为："+jedis.smembers("eleSet"));
        System.out.println("随机的移除集合中的一个元素："+jedis.spop("eleSet"));
        System.out.println("随机的移除集合中的一个元素："+jedis.spop("eleSet"));
        System.out.println("eleSet的所有元素为："+jedis.smembers("eleSet"));
        System.out.println("eleSet中包含元素的个数："+jedis.scard("eleSet"));
        System.out.println("e3是否在eleSet中："+jedis.sismember("eleSet", "e3"));
        System.out.println("e1是否在eleSet中："+jedis.sismember("eleSet", "e1"));
        System.out.println("e1是否在eleSet中："+jedis.sismember("eleSet", "e5"));
        System.out.println("=================================");
        System.out.println(jedis.sadd("eleSet1", "e1","e2","e4","e3","e0","e8","e7","e5"));
        System.out.println(jedis.sadd("eleSet2", "e1","e2","e4","e3","e0","e8"));
        System.out.println("将eleSet1中删除e1并存入eleSet3中："+jedis.smove("eleSet1", "eleSet3", "e1"));//移到集合元素
        System.out.println("将eleSet1中删除e2并存入eleSet3中："+jedis.smove("eleSet1", "eleSet3", "e2"));
        System.out.println("eleSet1中的元素："+jedis.smembers("eleSet1"));
        System.out.println("eleSet3中的元素："+jedis.smembers("eleSet3"));
        System.out.println("============集合运算=================");
        System.out.println("eleSet1中的元素："+jedis.smembers("eleSet1"));
        System.out.println("eleSet2中的元素："+jedis.smembers("eleSet2"));
        System.out.println("eleSet1和eleSet2的交集:"+jedis.sinter("eleSet1","eleSet2"));
        System.out.println("eleSet1和eleSet2的并集:"+jedis.sunion("eleSet1","eleSet2"));
        System.out.println("eleSet1和eleSet2的差集:"+jedis.sdiff("eleSet1","eleSet2"));//eleSet1中有，eleSet2中没有
        jedis.sinterstore("eleSet4","eleSet1","eleSet2");//求交集并将交集保存到dstkey的集合
        System.out.println("eleSet4中的元素："+jedis.smembers("eleSet4"));
    }
}

import redis.clients.jedis.Jedis;

public class TestList {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        jedis.flushDB();
        System.out.println("===========添加一个list===========");
        jedis.lpush("collections", "ArrayList", "Vector", "Stack", "HashMap", "WeakHashMap", "LinkedHashMap");
        jedis.lpush("collections", "HashSet");
        jedis.lpush("collections", "TreeSet");
        jedis.lpush("collections", "TreeMap");
        System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));//-1代表倒数第一个元素，-2代表倒数第二个元素,end为-1表示查询全部
        System.out.println("collections区间0-3的元素："+jedis.lrange("collections",0,3));
        System.out.println("===============================");
        // 删除列表指定的值 ，第二个参数为删除的个数（有重复时），后add进去的值先被删，类似于出栈
        System.out.println("删除指定元素个数："+jedis.lrem("collections", 2, "HashMap"));
        System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));
        System.out.println("删除下表0-3区间之外的元素："+jedis.ltrim("collections", 0, 3));
        System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));
        System.out.println("collections列表出栈（左端）："+jedis.lpop("collections"));
        System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));
        System.out.println("collections添加元素，从列表右端，与lpush相对应："+jedis.rpush("collections", "EnumMap"));
        System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));
        System.out.println("collections列表出栈（右端）："+jedis.rpop("collections"));
        System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));
        System.out.println("修改collections指定下标1的内容："+jedis.lset("collections", 1, "LinkedArrayList"));
        System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));
        System.out.println("===============================");
        System.out.println("collections的长度："+jedis.llen("collections"));
        System.out.println("获取collections下标为2的元素："+jedis.lindex("collections", 2));
        System.out.println("===============================");
        jedis.lpush("sortedList", "3","6","2","0","7","4");
        System.out.println("sortedList排序前："+jedis.lrange("sortedList", 0, -1));
        System.out.println(jedis.sort("sortedList"));
        System.out.println("sortedList排序后："+jedis.lrange("sortedList", 0, -1));
    }
}

import redis.clients.jedis.Jedis;
import java.util.Set;

public class TestKey {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1", 6379);

        System.out.println("清空数据："+jedis.flushDB());
        System.out.println("判断某个键是否存在："+jedis.exists("username"));
        System.out.println("新增<'username','kuangshen'>的键值对："+jedis.set("username", "kuangshen"));
        System.out.println("新增<'password','password'>的键值对："+jedis.set("password", "password"));
        System.out.print("系统中所有的键如下：");
        Set<String> keys = jedis.keys("*");
        System.out.println(keys);
        System.out.println("删除键password:"+jedis.del("password"));
        System.out.println("判断键password是否存在："+jedis.exists("password"));
        System.out.println("查看键username所存储的值的类型："+jedis.type("username"));
        System.out.println("随机返回key空间的一个："+jedis.randomKey());
        System.out.println("重命名key："+jedis.rename("username","name"));
        System.out.println("取出改后的name："+jedis.get("name"));
        System.out.println("按索引查询："+jedis.select(0));
        System.out.println("删除当前选择数据库中的所有key："+jedis.flushDB());
        System.out.println("返回当前数据库中key的数目："+jedis.dbSize());
        System.out.println("删除所有数据库中的所有key："+jedis.flushAll());
    }
}

import com.alibaba.fastjson.JSONObject;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;

public class TestMulti {
    public static void main(String[] args) {
        //创建客户端连接服务端，redis服务端需要被开启
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        jedis.flushDB();

        JSONObject jsonObject = new JSONObject();
        jsonObject.put("hello", "world");
        jsonObject.put("name", "java");
        //开启事务
        Transaction multi = jedis.multi();
        String result = jsonObject.toJSONString();
        try{
            //向redis存入一条数据
            multi.set("json", result);
            //再存入一条数据
            multi.set("json2", result);
            //这里引发了异常，用0作为被除数
            int i = 100/0;
            //如果没有引发异常，执行进入队列的命令
            multi.exec();
        }catch(Exception e){
            e.printStackTrace();
            //如果出现异常，回滚
            multi.discard();
        }finally{
            System.out.println(jedis.get("json"));
            System.out.println(jedis.get("json2"));
            //最终关闭客户端
            jedis.close();
        }
    }
}

七、springboot整合redis

jedis：采用的直连，多个线程操作的话，是不安全的，如果要避免不安全的，使用jedis pool连接池

lettuce：采用netty，实例可以再多个线程中进行共享，不存在不安全的情况

1、依赖

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

以2.5.5为例

源码：

源码分析：

@Configuration(
    proxyBeanMethods = false
)
@ConditionalOnClass({RedisOperations.class})
@EnableConfigurationProperties({RedisProperties.class})
@Import({LettuceConnectionConfiguration.class, JedisConnectionConfiguration.class})
public class RedisAutoConfiguration {
    public RedisAutoConfiguration() {
    }

    @Bean
    @ConditionalOnMissingBean(
        name = {"redisTemplate"}
    )
    @ConditionalOnSingleCandidate(RedisConnectionFactory.class)//可以自定义一个redisTemplate替换默认的
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
//默认的没有过多的设置，redis对象都是需要序列化的//两个类型都是oobject类型，根据需要转成<String,Object>        RedisTemplate<Object, Object> template = new RedisTemplate();
        template.setConnectionFactory(redisConnectionFactory);
        return template;
    }

    @Bean
    @ConditionalOnMissingBean //由于string是常用的类型，所以单独提出了一个
    @ConditionalOnSingleCandidate(RedisConnectionFactory.class)
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
        StringRedisTemplate template = new StringRedisTemplate();
        template.setConnectionFactory(redisConnectionFactory);
        return template;
    }
}

2、配置类application.propeties

//地址
spring.redis.host=xxx.x.x.x
//端口
spring.redis.port=xxxx
//密码
spring.redis.password=xxxxx
//库个数
spring.redis.database=xx

默认使用lettce

3、配置

@Configuration
public class RedisConfig {

    @Bean
    @SuppressWarnings("all")
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        //新建redisTemplate对象
        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
        //设置连接工厂
        template.setConnectionFactory(factory);
        //json序列化
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        //转义
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        //String的序列化
        jackson2JsonRedisSerializer.setObjectMapper(om);
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        
        // key采用String的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        template.setHashKeySerializer(stringRedisSerializer);
        template.setValueSerializer(jackson2JsonRedisSerializer);
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();
        return template;
    }
}

4、测试

指路：

一些常用的封装方法

八、配置文件 redis.conf

指路 https://www.cnblogs.com/shirleyxueli/articles/16921127.html

九、redis持久化

redis时候内存数据库，如果不将内存中的数据库状态保存到磁盘，一旦服务器进程退出，服务器中的数据库状态也会消失

1、RDB（Redis DataBase）

指定时间间隔内将内存中的数据集快照写入磁盘，也就是snapshot快照，恢复时将快照文件直接读到内存里。

redis会单独创建(一个fork)一个子进程来进行持久化，先将数据写入到一个临时文件，待持久化过程结束，再用这个临时文件替换上次持久化好的文件。整个过程，主进程是不进行任何IO操作的，确保了极高的性能。如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那RDB方式比AOF方式更加高效。RDB的缺点是最后一次持久化的数据可能丢失。

1.1 Fork

（1）作用：复制一个与当前进程一样的进程，新进程的所有数据（变量、环境变量、程序计数器等）数值都和原进程一致，但是是一个全新的进程，并作为原进程的子进程

（2）在 Linux 程序中，fork() 会产生一个和父进程完全相同的子进程，但子进程在此后多会 exec 系统调用，出于效率考虑，Linux 中引入写时复制技术

（3）一般情况父进程、子进程会共用同一段物理内存，只有进程空间的各段的内容要发生变化时，才会将父进程的内容复制一份给子进程

rdb保存的文件是dump.rdb

# 服务器在900秒内，对数据库进行了至少1次修改

save 900 1

# 服务器在300秒内，对数据库进行了至少10次修改

save 300 10

# 服务器在60秒内，对数据库进行了至少10000次修改

save 60 10000

# bgsave发生错误时是否停止写入，一般为yes

stop-writes-on-bgsave-error yes

# 持久化时是否使用LZF压缩字符串对象?

rdbcompression yes

# 是否对rdb文件进行校验和检验，通常为yes

rdbchecksum yes

# RDB持久化文件名

dbfilename dump.rdb

# 持久化文件存储目录

dir ./

1.2 触发机制

save的规则满足的情况

执行flushall

退出redis

1.3 恢复rdb文件

将rdb文件放在redis启动目录，redis启动的时候自动检查dump.rdb文件并恢复

查看需要存在的位置

127.0.0.1:6379> config get dir
1) "dir"
2) "/data"

1.4 优点及缺点

优点：

1、适合大规模的数据恢复

2、对数据完整性、一致性要求不高更适用

3、节省磁盘空间

4、恢复速度快

缺点：

需要一定的时间间隔进行操作，如果redis宕机，最后一次修改数据就没有了

fork进程的时候，占用一定的内存空间

2. AOF（Append Only File）

以日志的形式来记录每个写操作，将redis执行过的所有指令记录下来，只许追加文件但不可以改写文件，redis启动会读取该文件重新构建数据，就是说，redis重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作。

持久化流程：

（1）客户端的请求写命令会被 append 到 AOF 缓冲区内

（2）AOF 缓冲区根据 AOF 持久化策略，将操作 sync 到磁盘的 AOF 文件中

（3）AOF 文件大小超过重写策略或手动重写时，会对 AOF 文件 rewrite，压缩 AOF 文件容量

（4）Redis 服务重启时，会重新 load AOF 文件中的写操作，达到数据恢复的目的

aof保存的文件 appendonly.aof

默认是不开启的，需要手动配置

appendonly yes

如果aof有错误，redis无法启动，需要修复aof文件

redis-check-aof --fix appendonly.aof

2.1 优点及缺点

优点：

1、备份机制更稳健，丢失数据概率更低

2、可读的日志文本，通过操作 AOF 文件，可以处理误操作

appendonly no 　　　　　　　　　　#默认不开启

appendfilename "appendonlu.aof" 　　　#持久化文件的名字

# appendfsync always 　　　　　　 #每次修改都会sync，消耗性能

appendsync everysec 　　　　　　　　#每秒执行一次sync，可能会丢失这1s的数据

# appendfsync no 　　　　　　　　　 #不执行sync，这时候操作系统自己同步数据，速度最快

缺点：

1、相比 RDB，占用更多的磁盘空间

2、恢复备份速度要慢

3、每次读写都同步，有一定的性能压力

4、存在个别 Bug，造成不能恢复

2.2 重写规则

如果aof文件大于64MB，fork一个新的进程将文件重写

Redis 可以在 AOF 文件体积变得过大时，自动地在后台对 AOF 进行重写：重写后的新 AOF 文件包含了恢复当前数据集所需的最小命令集合。

整个重写操作是绝对安全的，因为 Redis 在创建新 AOF 文件的过程中，会继续将命令追加到现有的 AOF 文件里面，即使重写过程中发生停机，现有的 AOF 文件也不会丢失。

而一旦新 AOF 文件创建完毕，Redis 就会从旧 AOF 文件切换到新 AOF 文件，并开始对新 AOF 文件进行追加操作。

重写流程：

（1）bgrewriteaof 触发重写，判断是否当前有 bgsave 或 bgrewriteaof 在运行，如果有，则等待该命令结束后再继续执行

（2）主进程 fork 出子进程执行重写操作，保证主进程不会阻塞

（3）子进程遍历 Redis 内存中数据到临时文件，客户端的写请求同时写入 aof_buf（缓冲区）、aof_rewrite_buf（重写缓冲区），保证原 AOF 文件完整，以及新 AOF 文件生成期间，新的数据修改动作不会丢失

（4）子进程写完新的 AOF 文件后，向主进程发信号，父进程更新统计信息，主进程把 aof_rewrite_buf 中的数据写入到新的 AOF 文件

（5）使用新的 AOF 文件覆盖旧的 AOF 文件，完成 AOF 重写

2.3 扩展

选择持久化

1、如果只希望数据在服务器运行时存在，可以不使用任何持久化方式，即只做缓存

2、如果对数据不敏感，可以选单独使用 RDB

3、同时开启两种持久化方式

（1）在这种情况下，当 Redis 重启时会优先载入 AOF 文件来恢复原始的数据，因为在通常情况下 AOF 文件保存的数据集，要比 RDB 文件保存的数据集要完整

（2）RDB 数据不实时，同时使用两者时，服务器重启也只会找 AOF 文件

4、建议不只使用 AOF

（1）因为 RDB 更适合用于备份数据库，AOF 不断变化不容易备份

（2）RDB 适合快速重启，而且不会有 AOF 可能潜在 bug

5、性能建议

（1）RDB 文件只用作后备用途，建议只在 Slave 上持久化 RDB 文件，只保留 save 900 1

（2）如果使用 AOF，在最恶劣情况下也只会丢失不超过两秒数据

（3）AOF 代价：持续 I/O，rewrite 过程中产生的新数据，写到新文件造成的阻塞几乎是不可避免的

（4）建议 AOF 重写的基础大小设置到 5GB 以上，超过原大小 100% 大小时重写，可以改到适当的数值

十、redis发布订阅

常用命令：

序号	命令及描述
1	PSUBSCRIBE pattern [pattern ...] 订阅一个或多个符合给定模式的频道。
2	PUBSUB subcommand [argument [argument ...]] 查看订阅与发布系统状态。
3	PUBLISH channel message 将信息发送到指定的频道。
4	PUNSUBSCRIBE [pattern [pattern ...]] 退订所有给定模式的频道。
5	SUBSCRIBE channel [channel ...] 订阅给定的一个或多个频道的信息。
6	UNSUBSCRIBE [channel [channel ...]] 指退订给定的频道

10.1 原理

通过SUBSCRIBE命令订阅某个频道后，redis-server里面维护了一个字典，字典的键就是一个个channel，而字典的值则是一个链表，链表中保存了所有订阅这个channel的客户端。SUBSCRIBE命令的关键，就是将客户端添加到指定channel的订阅链表中。
通过PUBLISH命令向订阅者发送信息，redis-server会使用给定的频道作为键，在他所维护的channel字典中查找记录订阅该频道的所有客户端的链表，通过遍历这个链表，来将信息发布给所有的订阅者

十一、主从复制

11.1 概述

主从复制，是指将一台 Redis 服务器的数据，复制到其他的 Redis 服务器。前者称为主节点（Master/Leader）,后者称为从节点（Slave/Follower），数据的复制是单向的！只能由主节点复制到从节点（主节点以写为主、从节点以读为主）—— 读写分离。

11.2 作用

数据冗余：主从复制实现了数据的热备份，是持久化之外的一种数据冗余的方式。
故障恢复：当主节点故障时，从节点可以暂时替代主节点提供服务，是一种服务冗余的方式
负载均衡：在主从复制的基础上，配合读写分离，由主节点进行写操作，从节点进行读操作，分担服务器的负载；尤其是在多读少写的场景下，通过多个从节点分担负载，提高并发量。
高可用（集群）基石：主从复制还是哨兵和集群能够实施的基础。

一般来说，只使用一台redis是不能的

（1）从结构上，单个redis服务器会发生单点故障，并且一台服务器需要处理所有的请求负载，压力大。

（2）从容量上，单个redis服务器内存容量有限，就算一台redis内存容量为256G，也不能将所有内存用作redis存储内存，一般来说，单台redis最大使用内存不应该超过20G。

没有哨兵的时候，主机断了，从机依旧连接到主机，但是没有写操作，如果主机重新连上，从机依旧可以直接获取到主机写的信息。

如果使用命令行配置的主从，从机这个时候如果重启了，就变回主机。

11.3 复制原理

从机启动成功连接到主机后会发送一个sync命令

主机接到命令，启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令，在后台进程执行完毕后，主机将传送整个数据文件到从机，并完成一次完全同步。

全量复制：从机服务在接收到数据库文件数据后，将其存盘并加载到内存

增量复制：主机继续将新的所有手机到的修改命令一次传给从机，完成同步

但只要重新连接master，一次完全同步（全量）将被自动执行

11.4 宕机

如果主机断了，手动将从机设置为主机

slaveof no one

如果原来的主机重新连接，但是这个主机已经独立出来了

11.5 搭建

（1）查看当前库的信息

info replication

（2）配置多个redis.conf文件

这里以windows为例

redis79.conf：

port 6379　　　　　　　　#端口
daemonize yes　　　　　 #是否后台运行
pidfile /var/run/redis_6379.pid
logfile "6379.log"
dbfilename dump_6379.rdb

同理改其他的从配置

启动

redis-server redis79.conf
redis-server redis80.conf
redis-server redis81.conf

查看

ps -ef|grep redis

从机配置

 slaveof 172.17.0.3 6379

查看

info replication

真实的配置，应该是文件中配置：

redis80.conf

# replicaof <masterip> <masterport>
# masterauth <master-password>

拓展一个docker的：

docker run -itd --name redis-master -p 6379:6379 redis:latest
docker run -itd --name redis-slave-1 -p 6380:6380 redis:latest
docker run -itd --name redis-slave-2 -p 6381:6381 redis:latest

docker ps

查看redis-master的容器信息，找到IPAddress地址

docker inspect redis-master

进入redis-slave-1容器文件中

docker exec -it redis-slave-1 /bin/bash

进入redis-slave-1的密令行界面

redis-cli

指定当前Redis服务器为redis-master的从服务器。该命令的格式是 slaveof IP端口号，这里指向 172.17.0.2:6379 所在的主服务器

 slaveof 172.17.0.3 6379

两次退出

exit

同理第二个从节点配置同上

验证：

docker exec -it redis-master /bin/bash
redis-cli
info replication

docker的配置文件：

version: '3.7'
services:
  master:
    image: redis
    container_name: redis-master
    restart: always
    command: redis-server --requirepass redispwd  --appendonly yes
    ports:
      - 6379:6379
    volumes:
      - ./data1:/data
  slave1:
    image: redis
    container_name: redis-slave-1
    restart: always
    command: redis-server --slaveof redis-master 6379  --requirepass redispwd --masterauth redispwd  --appendonly yes
    ports:
      - 6380:6379
    volumes:
      - ./data2:/data
  slave2:
    image: redis
    container_name: redis-slave-2
    restart: always
    command: redis-server --slaveof redis-master 6379  --requirepass redispwd --masterauth redispwd  --appendonly yes
    ports:
      - 6381:6379
    volumes:
      - ./data3:/data

创建文件夹：

mkdir redis
cd redis
vim docker-compose.yml

进入docker-compose.yml所在的文件夹中，执行命令

docker-compose up -d

十二、哨兵模式

redis从2.8开始正式提供了sentinel架构来解决手动干预切换主服务器的方式。

哨兵模式是一种特殊的模式，首先redis提供了哨兵的命令，哨兵是一个独立的进程，作为进程会独立运行，原理是哨兵通过发送命令，等待redis服务器响应，从而监控运行的多个redis实例。

12.1 作用

通过发送命令，让redis服务器返回监控其运行状态，包括主服务器和从服务器
当哨兵检测到主机宕机，会自动将从机切换成主机，然后通过发布订阅模式通知其他的从服务器，修改配置文件，让他们切换主机

然而一个哨兵进程对redis服务器进行监控，可能会出现问题，所以可以使用多个哨兵进行监控，各个哨兵之间还会进行监控，就形成了多哨兵模式

假设主服务器宕机，哨兵1先检测到这个结果，系统并不会马上进行failover过程，仅仅是哨兵1主观的认为主服务器不可用，这个现象称为-主观下线。

当后面的哨兵也检测到主服务器不可用，并且数量达到一定值，哨兵之间就会进行一次投票，投票的结果由一个哨兵发起，进行failover故障转移操作，切换成功后，通过发布订阅模式，让各个哨兵把自己监控的从服务器实现切换主机，这个过程称为-客观下线。

12.2 配置部署

（1）sentinel.conf

######sentinel monitor 被监控的名字 host port 1
######1代表主机挂了，从机投票看让谁成为主机
sentinel monitor redis-master 127.0.0.4 6379 1

全部的配置

# Example sentinel.conf
# 哨兵sentinel实例运行的端口 默认26379
port 26379

# 哨兵sentinel的工作目录
dir /tmp

# 哨兵sentinel监控的redis主节点的 ip port
# master-name 可以自己命名的主节点名字 只能由字母A-z、数字0-9 、这三个字符".-_"组成。
# quorum 配置多少个sentinel哨兵统一认为master主节点失联 那么这时客观上认为主节点失联了
# sentinel monitor <master-name> <ip> <redis-port> <quorum>
sentinel monitor mymaster 127.0.0.1 6379 2

# 当在Redis实例中开启了requirepass foobared 授权密码 这样所有连接Redis实例的客户端都要提供
密码
# 设置哨兵sentinel 连接主从的密码 注意必须为主从设置一样的验证密码
# sentinel auth-pass <master-name> <password>
sentinel auth-pass mymaster MySUPER--secret-0123passw0rd

# 指定多少毫秒之后 主节点没有应答哨兵sentinel 此时 哨兵主观上认为主节点下线 默认30秒
# sentinel down-after-milliseconds <master-name> <milliseconds>
sentinel down-after-milliseconds mymaster 30000

# 这个配置项指定了在发生failover主备切换时最多可以有多少个slave同时对新的master进行 同步，
# 这个数字越小，完成failover所需的时间就越长，
# 但是如果这个数字越大，就意味着越 多的slave因为replication而不可用。
# 可以通过将这个值设为 1 来保证每次只有一个slave 处于不能处理命令请求的状态。
# sentinel parallel-syncs <master-name> <numslaves>
sentinel parallel-syncs mymaster 1

# 故障转移的超时时间 failover-timeout 可以用在以下这些方面：
#1. 同一个sentinel对同一个master两次failover之间的间隔时间。
#2. 当一个slave从一个错误的master那里同步数据开始计算时间。直到slave被纠正为向正确的master那里同步数据时。
#3.当想要取消一个正在进行的failover所需要的时间。
#4.当进行failover时，配置所有slaves指向新的master所需的最大时间。不过，即使过了这个超时，slaves依然会被正确配置为指向master，但是就不按parallel-syncs所配置的规则来了
# 默认三分钟
# sentinel failover-timeout <master-name> <milliseconds>

# Redis缓存的使用，极大的提升了应用程序的性能和效率，特别是数据查询方面。但同时，它也带来了一些问题。其中，最要害的问题，就是数据的一致性问题，从严格意义上讲，这个问题无解。如果对数据的一致性要求很高，那么就不能使用缓存。另外的一些典型问题就是，缓存穿透、缓存雪崩和缓存击穿。目前，业界也都有比较流行的解决方案。
sentinel failover-timeout mymaster 180000

# SCRIPTS EXECUTION
#配置当某一事件发生时所需要执行的脚本，可以通过脚本来通知管理员，例如当系统运行不正常时发邮件通知相关人员。
#对于脚本的运行结果有以下规则：
#若脚本执行后返回1，那么该脚本稍后将会被再次执行，重复次数目前默认为10
#若脚本执行后返回2，或者比2更高的一个返回值，脚本将不会重复执行。
#如果脚本在执行过程中由于收到系统中断信号被终止了，则同返回值为1时的行为相同。
#一个脚本的最大执行时间为60s，如果超过这个时间，脚本将会被一个SIGKILL信号终止，之后重新执行。
#通知型脚本:当sentinel有任何警告级别的事件发生时（比如说redis实例的主观失效和客观失效等等），将会去调用这个脚本，这时这个脚本应该通过邮件，SMS等方式去通知系统管理员关于系统不正常运行的信息。
#调用该脚本时，将传给脚本两个参数，一个是事件的类型，一个是事件的描述。如果sentinel.conf配置文件中配置了这个脚本路径，那么必须保证这个脚本存在于这个路径，并且是可执行的，否则sentinel无法正常启动成功。
#通知脚本
# shell编程
# sentinel notification-script <master-name> <script-path>
sentinel notification-script mymaster /var/redis/notify.sh

# 客户端重新配置主节点参数脚本
# 当一个master由于failover而发生改变时，这个脚本将会被调用，通知相关的客户端关于master地址已经发生改变的信息。
# 以下参数将会在调用脚本时传给脚本:
# <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port>
# 目前<state>总是“failover”,
# <role>是“leader”或者“observer”中的一个。
# 参数 from-ip, from-port, to-ip, to-port是用来和旧的master和新的master(即旧的slave)通信的
# 这个脚本应该是通用的，能被多次调用，不是针对性的。
# sentinel client-reconfig-script <master-name> <script-path>
sentinel client-reconfig-script mymaster /var/redis/reconfig.sh # 一般都是由运维来配

（2）启动