1.REDIS
2. redis介绍
端口 6379 ， 来自一个女歌手
默认16个数据库，类似
使用命令 select id 来切换数据库。如: select 8
统一密码管理，所有库同样密码。
Copy dbsize
flushdb
flushall
123 redis底层使用的是：
Redis是单线程+多路IO复用技术
3. 常用的五大基本数据类型
Copy 归纳：
1. String
2. List
3. set
4. sort set
5. hash
123456 关于hash不是熟悉的话，回去复习数据结构。
1. 对key的基本操作
Copy keys * 查看当前库所有key (匹配：keys *1)
1
Copy exists key 判断某个key是否存在
1
Copy type key 查看你的key是什么类型
1
Copy del key 删除指定的key数据
unlink key 根据value选择非阻塞删除
仅将keys从keyspace元数据中删除，真正的删除会在后续异步操作。
123
Copy expire key 10 10秒钟：为给定的key设置过期时间
ttl key 查看还有多少秒过期，-1表示永不过期，-2表示已过期
12
Copy select命令切换数据库
dbsize查看当前数据库的key的数量
flushdb清空当前库
flushall通杀全部库
1234 Redis字符串（String）
Copy String是Redis最基本的类型，你可以理解成与Memcached一模一样的类型，一个key对应一个value。
String类型是二进制安全的。意味着Redis的string可以包含任何数据。比如jpg图片或者序列化的对象。
String类型是Redis最基本的数据类型，一个Redis中字符串value最多可以是512M
123
Copy set <key><value>添加键值对
get <key>查询对应键值
append <key><value>将给定的<value> 追加到原值的末尾，返回字符串的长度
strlen <key>获得值的长度
setnx <key><value>只有在 key 不存在时 设置 key 的值
incr <key>
将 key 中储存的数字值增1
只能对数字值操作，如果为空，新增值为1
decr <key>
将 key 中储存的数字值减1
只能对数字值操作，如果为空，新增值为-1
incrby / decrby <key><步长>将 key 中储存的数字值增减。自定义步长。
12345678910111213141516
Copy mset <key1><value1><key2><value2> .....
同时设置一个或多个 key-value对
mget <key1><key2><key3> .....
同时获取一个或多个 value
msetnx <key1><value1><key2><value2> .....
同时设置一个或多个 key-value 对，当且仅当所有给定 key 都不存在。
注：原子性，有一个失败则都失败
1234567891011
Copy getrange <key><起始位置><结束位置>
获得值的范围，类似java中的substring，前包，后包
setrange <key><起始位置><value>
用 <value> 覆写<key>所储存的字符串值，从<起始位置>开始(索引从0开始)。
12345
Copy setex <key><过期时间><value>
设置键值的同时，设置过期时间，单位秒。
getset <key><value>
以新换旧，设置了新值同时获得旧值。
12345
Copy 这个数据就是：动态数组，java中的BufferString ,可以这么理解
1 Redis列表（List）
单键多值 Redis 列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部（左边）或者尾部（右边）。
Copy lpush/rpush <key><value1><value2><value3> .... 从左边/右边插入一个或多个值。
lpop/rpop <key> 从左边/右边吐出一个值。值在键在，值光键亡。
rpoplpush <key1> <key2> 从<key1>列表右边吐出一个值，插到<key2>列表左边。
lrange <key> <start> <stop>
按照索引下标获得元素(从左到右)
lrange mylist 0 -1 0左边第一个，-1右边第一个，（0-1表示获取所有）
lindex <key><index> 按照索引下标获得元素(从左到右)
llen <key> 获得列表长度
linsert <key> before <value> <newvalue> 在<value>的前面插入<newvalue>插入值
linsert <key> after <value> <newvalue> 在<value>的后面插入<newvalue>插入值
lrem <key><n><value> 从左边删除n个value(从左到右)
lset<key><index><value> 将列表key下标为index的值替换成value
123456789101112131415161718192021 List的数据结构为快速链表quickList
Copy 列表元素较少的情况下会使用一块连续的内存存储，这个结构是ziplist，也即是压缩列表。
元素多了，就使用 多个zipList进行存储，用双向链表的形式，连接
12 Redis集合（Set）
Copy set集合中的元素 是无序的，存取无序，且元素不能重复。
1 Redis的Set是string类型的无序集合。它底层其实是一个value为null的hash表 ，所以添加，删除，查找的复杂度都是O(1)。
Copy sadd <key><value1><value2> ..... 将一个或多个 member 元素加入到集合 key 中，已经存在的 member 元素将被忽略
smembers <key> 取出该集合的所有值。
sismember <key><value> 判断集合<key>是否为含有该<value>值，有1，没有0
scard<key> 返回该集合的元素个数。
srem <key><value1><value2> .... 删除集合中的某个元素。
spop <key> 随机从该集合中吐出一个值。
srandmember <key><n> 随机从该集合中取出n个值。不会从集合中删除 。
smove <source><destination>value 把集合中一个值从一个集合移动到另一个集合
sinter <key1><key2> 返回两个集合的交集元素。
sunion <key1><key2>返回两个集合的并集元素。
sdiff <key1><key2>返回两个集合的差集元素(key1中的，不包含key2中的)
12345678910111213 Redis哈希（Hash）
Redis hash 是一个键值对集合。 Redis hash是一个string类型的field和value的映射表，hash特别适合用于存储对象。 类似Java里面的Map<String,Object> 用户ID为查找的key，存储的value用户对象包含姓名，年龄，生日等信息，如果用普通的key/value结构来存储
Copy hset <key><field><value>给<key>集合中的 <field>键赋值<value>
hget <key1><field>从<key1>集合<field>取出 value
hmset <key1><field1><value1><field2><value2>... 批量设置hash的值
hexists<key1><field>查看哈希表 key 中，给定域 field 是否存在。
hkeys <key>列出该hash集合的所有field
hvals <key>列出该hash集合的所有value
hincrby <key><field><increment>为哈希表 key 中的域 field 的值加上增量 1 -1
hsetnx <key><field><value>将哈希表 key 中的域 field 的值设置为 value ，当且仅当域 field 不存在 .
12345678 Redis有序集合Zset(sorted set)
Copy zadd <key><score1><value1><score2><value2>…
将一个或多个 member 元素及其 score 值加入到有序集 key 当中。
zrange <key><start><stop> [WITHSCORES]
返回有序集 key 中，下标在<start><stop>之间的元素
带WITHSCORES，可以让分数一起和值返回到结果集。
zrange key 0 -1 取出所有，不包括分数
zrange key 0 -1 withscores 取出所有，包括分数
zrangebyscore key minmax [withscores] [limit offset count]
返回有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员。有序集成员按 score 值递增(从小到大)次序排列。
zrevrangebyscore key maxmin [withscores] [limit offset count]
同上，改为从大到小排列。
zincrby <key><increment><value> 为元素的score加上增量
zrem <key><value>删除该集合下，指定值的元素
zcount <key><min><max>统计该集合，分数区间内的元素个数
zrank <key><value>返回该值在集合中的排名，从0开始。
123456789101112131415161718192021 **SortedSet(zset)**是Redis提供的一个非常特别的数据结构，
一方面它等价于Java的数据结构Map<String, Double>，可以给每一个元素value赋予一个权重score，另一方面它又类似于
TreeSet，内部的元素会按照权重score进行排序，可以得到每个元素的名次，还可以通过score的范围来获取元素的列表。
Copy 其实就是排序中的 最优树。查找快，本来也就是通过有序进行排序的。
1 4. Redis配置文件
Copy 打开 redis文件中的 redis.conf 文件
1 文件模块：
###Units单位###
配置大小单位,开头定义了一些基本的度量单位，只支持bytes，不支持bit 大小写不敏感
###INCLUDES包含###
这个可以包含其他的conf文件。
类似jsp中的include，多实例的情况可以把公用的配置文件提取出来
###网络相关配置
1. bind
默认情况bind=127.0.0.1只能接受本机的访问请求
不写的情况下，无限制接受任何ip地址的访问
生产环境肯定要写你应用服务器的地址；服务器是需要远程访问的，所以需要将其注释掉
如果开启了protected-mode ，那么在没有设定bind ip且没有设密码的情况下，Redis只允许接受本机的响应
保存配置，停止服务，重启启动查看进程，不再是本机访问了。
2. protected-mode
将本机访问保护模式设置no, 远程就可以连接了。
3. Port
端口号，默认 6379
4.tcp-backlog
设置tcp的backlog，backlog其实是一个连接队列，backlog队列总和=未完成三次握手队列 + 已经完成三次握手队列。 在高并发环境下你需要一个高backlog值来避免慢客户端连接问题。
注意Linux内核会将这个值减小到/proc/sys/net/core/somaxconn的值（128），所以需要确认增大/proc/sys/net/core/somaxconn和/proc/sys/net/ipv4/tcp_max_syn_backlog（128）两个值来达到想要的效果
5. timeout
一个空闲的客户端维持多少秒会关闭，0表示关闭该功能。即永不关闭。
6. tcp-keepalive
对访问客户端的一种心跳检测，每个n秒检测一次。
单位为秒，如果设置为0，则不会进行Keepalive检测，建议设置成60
###GENERAL通用###
1. daemonize
是否为后台进程，设置为yes
2. pidfile
存放pid文件的位置，每个实例会产生一个不同的pid文件，存放进程号的。
一个程序可以有多个进程，一个进程对应一个端口号，进程号。
3. loglevel
指定日志记录级别，Redis总共支持四个级别：debug、verbose、notice、warning，默认为notice
四个级别根据使用阶段来选择，生产环境选择notice 或者warning
4. logfile
5. databases 16
###SECURITY安全###
1. 设置密码
在命令中设置密码，只是临时的。重启redis服务器，密码就还原了。
永久设置，需要再配置文件中进行设置。
#### LIMITS限制
1. maxclients
设置redis同时可以与多少个客户端进行连接。 默认情况下为10000个客户端。 如果达到了此限制，redis则会拒绝新的连接请求，并且向这些连接请求方发出“max number of clients reached”以作回应。
2. maxmemory
建议必须设置，否则，将内存占满，造成服务器宕机 设置redis可以使用的内存量。一旦到达内存使用上限，redis将会试图移除内部数据，移除规则可以通过maxmemory-policy来指定。 如果redis无法根据移除规则来移除内存中的数据，或者设置了“不允许移除”，那么redis则会针对那些需要申请内存的指令返回错误信息，比如SET、LPUSH等。 但是对于无内存申请的指令，仍然会正常响应，比如GET等。如果你的redis是主redis（说明你的redis有从redis），那么在设置内存使用上限时，需要在系统中留出一些内存空间给同步队列缓存，只有在你设置的是“不移除”的情况下，才不用考虑这个因素。
3. maxmemory-policy
volatile-lru：使用LRU算法移除key，只对设置了过期时间的键；（最近最少使用） allkeys-lru：在所有集合key中，使用LRU算法移除key volatile-random：在过期集合中移除随机的key，只对设置了过期时间的键 allkeys-random：在所有集合key中，移除随机的key volatile-ttl：移除那些TTL值最小的key，即那些最近要过期的key noeviction：不进行移除。针对写操作，只是返回错误信息
4.maxmemory-samples
设置样本数量，LRU算法和最小TTL算法都并非是精确的算法，而是估算值，所以你可以设置样本的大小，redis默认会检查这么多个key并选择其中LRU的那个。 一般设置3到7的数字，数值越小样本越不准确，但性能消耗越小。
5. Redis的发布和订阅
1. 什么是发布和订阅
Redis 发布订阅 (pub/sub) 是一种消息通信模式：发送者 (pub) 发送消息，订阅者 (sub) 接收消息。
Redis 客户端可以订阅任意数量的频道。
2. Redis的发布和订阅
3. 发布订阅命令行实现
1、打开一个客户端订阅channel1
2、打开另一个客户端，给channel1发布消息hello
3、打开第一个客户端可以看到发送的消息
6.Redis新数据类型
1. Bitmaps
简介
命令
setbit （1）格式
Copy setbit<key><offset><value> 设置Bitmaps中某个偏移量的值（0或1）
1 2、getbit （1）格式
Copy getbit<key><offset>获取Bitmaps中某个偏移量的值
1 3、bitcount 统计字符串被设置为1的bit数。一般情况下，给定的整个字符串都会被进行计数，通过指定额外的 start 或 end 参数，可以让计数只在特定的位上进行。start 和 end 参数的设置，都可以使用负数值：比如 -1 表示最后一个位，而 -2 表示倒数第二个位，start、end 是指bit组的字节的下标数，二者皆包含。
（1）格式
Copy bitcount<key>[start end] 统计字符串从start字节到end字节比特值为1的数量
1 bitcount K1 1 3 ： 统计下标1、2字节组中bit=1的个数，即01000000 00000000 00100001 –》bitcount K1 1 3 --》3
bitcount K1 0 -2 ： 统计下标0到下标倒数第2，字节组中bit=1的个数，即 01000001 01000000 00000000 –》bitcount K1 0 -2 --》3
注意：redis的setbit设置或清除的是bit位置，而bitcount计算的是byte位置。
4、bitop (1)格式
Copy bitop and(or/not/xor) <destkey> [key…]
1 (2)实例 2020-11-04 日访问网站的userid=1,2,5,9。 setbit unique:users:20201104 1 1 setbit unique:users:20201104 2 1 setbit unique:users:20201104 5 1 setbit unique:users:20201104 9 1
2020-11-03 日访问网站的userid=0,1,4,9。 setbit unique:users:20201103 0 1 setbit unique:users:20201103 1 1 setbit unique:users:20201103 4 1 setbit unique:users:20201103 9 1
Bitmaps与set对比
但Bitmaps并不是万金油， 假如该网站每天的独立访问用户很少， 例如只有10万（大量的僵尸用户） ， 那么两者的对比如下表所示， 很显然， 这时候使用Bitmaps就不太合适了， 因为基本上大部分位都是0。
2. HyperLogLog
简介
在工作当中，我们经常会遇到与统计相关的功能需求，比如统计网站PV（PageView页面访问量）,可以使用Redis的incr、incrby轻松实现。 但像UV（UniqueVisitor，独立访客）、独立IP数、搜索记录数等需要去重和计数的问题如何解决？这种求集合中不重复元素个数的问题称为基数问题。 解决基数问题有很多种方案： （1）数据存储在MySQL表中，使用distinct count计算不重复个数 （2）使用Redis提供的hash、set、bitmaps等数据结构来处理 以上的方案结果精确，但随着数据不断增加，导致占用空间越来越大，对于非常大的数据集是不切实际的。 能否能够降低一定的精度来平衡存储空间？Redis推出了HyperLogLog Redis HyperLogLog 是用来做基数统计的算法，HyperLogLog 的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定的、并且是很小的。 在 Redis 里面，每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比。 但是，因为 HyperLogLog 只会根据输入元素来计算基数，而不会储存输入元素本身，所以 HyperLogLog 不能像集合那样，返回输入的各个元素。
什么是基数? 比如数据集 {1, 3, 5, 7, 5, 7, 8}， 那么这个数据集的基数集为 {1, 3, 5 ,7, 8}, 基数(不重复元素)为5。 基数估计就是在误差可接受的范围内，快速计算基数。
命令
3. Geospatial
简介
Redis 3.2 中增加了对GEO类型的支持。GEO，Geographic，地理信息的缩写。该类型，就是元素的2维坐标，在地图上就是经纬度。redis基于该类型，提供了经纬度设置，查询，范围查询，距离查询，经纬度Hash等常见操作。
命令
1、geoadd
3、geodist
7. Redis_Jedis_测试
7.1.Jedis所需要的jar包
Copy <dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.2.0</version>
</dependency>
12345 7.2.连接Redis注意事项
禁用Linux的防火墙：Linux(CentOS7)里执行命令 systemctl stop/disable firewalld.service redis.conf中注释掉bind 127.0.0.1 ,然后 protected-mode no
7.3.Jedis常用操作
创建测试程序
Copy package com.atguigu.jedis;
import redis.clients.jedis.Jedis;
public class Demo01 {
public static void main(String[] args) {
Jedis jedis = new Jedis("192.168.137.3",6379);
String pong = jedis.ping();
System.out.println("连接成功："+pong);
jedis.close();
}
}
12345678910 7.4.1.Jedis-API: Key
Copy jedis.set("k1", "v1");
jedis.set("k2", "v2");
jedis.set("k3", "v3");
Set<String> keys = jedis.keys("*");
System.out.println(keys.size());
for (String key : keys) {
System.out.println(key);
}
System.out.println(jedis.exists("k1"));
System.out.println(jedis.ttl("k1"));
System.out.println(jedis.get("k1"));
1234567891011 7.4.2.Jedis-API: String
Copy jedis.mset("str1","v1","str2","v2","str3","v3");
System.out.println(jedis.mget("str1","str2","str3"));
12 7.4.3.Jedis-API: List
Copy List<String> list = jedis.lrange("mylist",0,-1);
for (String element : list) {
System.out.println(element);
}
1234 7.4.4.Jedis-API: set
Copy jedis.sadd("orders", "order01");
jedis.sadd("orders", "order02");
jedis.sadd("orders", "order03");
jedis.sadd("orders", "order04");
Set<String> smembers = jedis.smembers("orders");
for (String order : smembers) {
System.out.println(order);
}
jedis.srem("orders", "order02");
123456789 7.4.5.Jedis-API: hash
Copy jedis.hset("hash1","userName","lisi");
System.out.println(jedis.hget("hash1","userName"));
Map<String,String> map = new HashMap<String,String>();
map.put("telphone","13810169999");
map.put("address","atguigu");
map.put("email","abc@163.com");
jedis.hmset("hash2",map);
List<String> result = jedis.hmget("hash2", "telphone","email");
for (String element : result) {
System.out.println(element);
}
1234567891011 7.4.6.Jedis-API: zset
Copy jedis.zadd("zset01", 100d, "z3");
jedis.zadd("zset01", 90d, "l4");
jedis.zadd("zset01", 80d, "w5");
jedis.zadd("zset01", 70d, "z6");
Set<String> zrange = jedis.zrange("zset01", 0, -1);
for (String e : zrange) {
System.out.println(e);
}
123456789 测试
Copy package com.atguigu.jedis;
import org.junit.Test;
import redis.clients.jedis.Jedis;
import java.util.List;
import java.util.Set;
public class JedisDemo1 {
public static void main(String[] args) {
//创建Jedis对象
Jedis jedis = new Jedis("192.168.44.168",6379);
//测试
String value = jedis.ping();
System.out.println(value);
jedis.close();
}
//操作zset
@Test
public void demo5() {
//创建Jedis对象
Jedis jedis = new Jedis("192.168.44.168",6379);
jedis.zadd("china",100d,"shanghai");
Set<String> china = jedis.zrange("china", 0, -1);
System.out.println(china);
jedis.close();
}
//操作hash
@Test
public void demo4() {
//创建Jedis对象
Jedis jedis = new Jedis("192.168.44.168",6379);
jedis.hset("users","age","20");
String hget = jedis.hget("users", "age");
System.out.println(hget);
jedis.close();
}
//操作set
@Test
public void demo3() {
//创建Jedis对象
Jedis jedis = new Jedis("192.168.44.168",6379);
jedis.sadd("names","lucy");
jedis.sadd("names","mary");
Set<String> names = jedis.smembers("names");
System.out.println(names);
jedis.close();
}
//操作list
@Test
public void demo2() {
//创建Jedis对象
Jedis jedis = new Jedis("192.168.44.168",6379);
jedis.lpush("key1","lucy","mary","jack");
List<String> values = jedis.lrange("key1", 0, -1);
System.out.println(values);
jedis.close();
}
//操作key string
@Test
public void demo1() {
//创建Jedis对象
Jedis jedis = new Jedis("192.168.44.168",6379);
//添加
jedis.set("name","lucy");
//获取
String name = jedis.get("name");
System.out.println(name);
//设置多个key-value
jedis.mset("k1","v1","k2","v2");
List<String> mget = jedis.mget("k1", "k2");
System.out.println(mget);
Set<String> keys = jedis.keys("*");
for(String key : keys) {
System.out.println(key);
}
jedis.close();
}
}
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697 8.Redis_Jedis_实例
8.1.完成一个手机验证码功能
要求： 1、输入手机号，点击发送后随机生成6位数字码，2分钟有效 2、输入验证码，点击验证，返回成功或失败 3、每个手机号每天只能输入3次
Copy package com.atguigu.jedis;
import redis.clients.jedis.Jedis;
import java.util.Random;
public class PhoneCode {
public static void main(String[] args) {
//模拟验证码发送
verifyCode("13678765435");
//模拟验证码校验
//getRedisCode("13678765435","4444");
}
//3 验证码校验
public static void getRedisCode(String phone,String code) {
//从redis获取验证码
Jedis jedis = new Jedis("192.168.44.168",6379);
//验证码key
String codeKey = "VerifyCode"+phone+":code";
String redisCode = jedis.get(codeKey);
//判断
if(redisCode.equals(code)) {
System.out.println("成功");
}else {
System.out.println("失败");
}
jedis.close();
}
//2 每个手机每天只能发送三次，验证码放到redis中，设置过期时间120
public static void verifyCode(String phone) {
//连接redis
Jedis jedis = new Jedis("192.168.44.168",6379);
//拼接key
//手机发送次数key
String countKey = "VerifyCode"+phone+":count";
//验证码key
String codeKey = "VerifyCode"+phone+":code";
//每个手机每天只能发送三次
String count = jedis.get(countKey);
if(count == null) {
//没有发送次数，第一次发送
//设置发送次数是1
jedis.setex(countKey,24*60*60,"1");
} else if(Integer.parseInt(count)<=2) {
//发送次数+1
jedis.incr(countKey);
} else if(Integer.parseInt(count)>2) {
//发送三次，不能再发送
System.out.println("今天发送次数已经超过三次");
jedis.close();
}
//发送验证码放到redis里面
String vcode = getCode();
jedis.setex(codeKey,120,vcode);
jedis.close();
}
//1 生成6位数字验证码
public static String getCode() {
Random random = new Random();
String code = "";
for(int i=0;i<6;i++) {
int rand = random.nextInt(10);
code += rand;
}
return code;
}
}
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576 9.Redis与Spring Boot整合
1、在pom.xml文件中引入redis相关依赖
Copy <!-- redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- spring2.X集成redis所需common-pool2-->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>2.6.0</version>
</dependency>
123456789101112 2、application.properties配置redis配置
Copy #Redis服务器地址
spring.redis.host=192.168.140.136
#Redis服务器连接端口
spring.redis.port=6379
#Redis数据库索引（默认为0）
spring.redis.database= 0
#连接超时时间（毫秒）
spring.redis.timeout=1800000
#连接池最大连接数（使用负值表示没有限制）
spring.redis.lettuce.pool.max-active=20
#最大阻塞等待时间(负数表示没限制)
spring.redis.lettuce.pool.max-wait=-1
#连接池中的最大空闲连接
spring.redis.lettuce.pool.max-idle=5
#连接池中的最小空闲连接
spring.redis.lettuce.pool.min-idle=0
12345678910111213141516 3、添加redis配置类
Copy @EnableCaching
@Configuration
public class RedisConfig extends CachingConfigurerSupport {
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
RedisSerializer<String> redisSerializer = new StringRedisSerializer();
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
template.setConnectionFactory(factory);
//key序列化方式
template.setKeySerializer(redisSerializer);
//value序列化
template.setValueSerializer(jackson2JsonRedisSerializer);
//value hashmap序列化
template.setHashValueSerializer(jackson2JsonRedisSerializer);
return template;
}
@Bean
public CacheManager cacheManager(RedisConnectionFactory factory) {
RedisSerializer<String> redisSerializer = new StringRedisSerializer();
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
//解决查询缓存转换异常的问题
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
// 配置序列化（解决乱码的问题）,过期时间600秒
RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
.entryTtl(Duration.ofSeconds(600))
.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))
.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
.disableCachingNullValues();
RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
.cacheDefaults(config)
.build();
return cacheManager;
}
}
1234567891011121314151617181920212223242526272829303132333435363738394041424344 4、测试一下 RedisTestController中添加测试方法
Copy @RestController
@RequestMapping("/redisTest")
public class RedisTestController {
@Autowired
private RedisTemplate redisTemplate;
@GetMapping
public String testRedis() {
//设置值到redis
redisTemplate.opsForValue().set("name","lucy");
//从redis获取值
String name = (String)redisTemplate.opsForValue().get("name");
return name;
}
}
123456789101112131415 10.Redis_事务_锁机制_秒杀
10.1.Redis的事务定义
Redis事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。
10.2.Multi、Exec、discard
从输入Multi命令开始，输入的命令都会依次进入命令队列中，但不会执行，直到输入Exec后，Redis会将之前的命令队列中的命令依次执行。
10.3.事务的错误处理
10.4 乐观锁和悲观锁
想想一个场景：有很多人有你的账户,同时去参加双十一抢购
10.5.2.悲观锁
加锁
10.5.3.乐观锁
加版本号
10.5.4.WATCH key [key …] 乐观锁的使用
在执行multi之前，先执行watch key1 [key2],可以监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。
10.5.5.unwatch
取消 WATCH 命令对所有 key 的监视。 如果在执行 WATCH 命令之后，EXEC 命令或DISCARD 命令先被执行了的话，那么就不需要再执行UNWATCH 了。 http://doc.redisfans.com/transaction/exec.html
Redis事务三特性
单独的隔离操作 事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。 没有隔离级别的概念 队列中的命令没有提交之前都不会实际被执行，因为事务提交前任何指令都不会被实际执行 不保证原子性 事务中如果有一条命令执行失败，其后的命令仍然会被执行，没有回滚
11.Redis_事务_秒杀案例(感受事务操作)
11.1.解决计数器和人员记录的事务操作
11.2.Redis事务–秒杀并发模拟
使用工具ab模拟测试 CentOS6 默认安装 CentOS7需要手动安装
11.2.2.无网络
（1） 进入cd /run/media/root/CentOS 7 x86_64/Packages（路径跟centos6不同） （2） 顺序安装 apr-1.4.8-3.el7.x86_64.rpm apr-util-1.5.2-6.el7.x86_64.rpm httpd-tools-2.4.6-67.el7.centos.x86_64.rpm
11.2.3.测试及结果
11.2.3.1.通过ab测试
vim postfile 模拟表单提交参数,以&符号结尾;存放当前目录。 内容：prodid=0101&
ab -n 2000 -c 200 -k -p ~/postfile -T application/x-www-form-urlencoded http://192.168.2.115:8081/Seckill/doseckill
-n 是指请求的次数 -c 是并发的次数 -T 如果是POst/put的方式请求，必须要设置content-type 类型 -p 指 请求的参数放在 指定的文件postfile文件中
代码中使用jedis连接池的技术，防止每次一个请求进行连接一个redis的过程，消耗性能，且可能会出现 连接超时的错误。
11.2.3.2.超卖
11.3.超卖问题
11.4.利用乐观锁淘汰用户，解决超卖问题。
Copy //增加乐观锁
jedis.watch(qtkey);
//3.判断库存
String qtkeystr = jedis.get(qtkey);
if(qtkeystr==null || "".equals(qtkeystr.trim())) {
System.out.println("未初始化库存");
jedis.close();
return false ;
}
int qt = Integer.parseInt(qtkeystr);
if(qt<=0) {
System.err.println("已经秒光");
jedis.close();
return false;
}
123456789101112131415161718
Copy //增加事务
Transaction multi = jedis.multi();
//4.减少库存
//jedis.decr(qtkey);
multi.decr(qtkey);
//5.加人
//jedis.sadd(usrkey, uid);
multi.sadd(usrkey, uid);
//执行事务
List<Object> list = multi.exec();
//判断事务提交是否失败
if(list==null || list.size()==0) {
System.out.println("秒杀失败");
jedis.close();
return false;
}
System.err.println("秒杀成功");
jedis.close();
12345678910111213141516171819202122 11.5.继续增加并发测试
11.5.1.连接有限制
ab -n 2000 -c 200 -k -p postfile -T ‘application/x-www-form-urlencoded’ http://192.168.140.1:8080/seckill/doseckill
增加-r参数，-r Don’t exit on socket receive errors. ab -n 2000 -c 100 -r -p postfile -T ‘application/x-www-form-urlencoded’ http://192.168.140.1:8080/seckill/doseckill
11.5.2.已经秒光，可是还有库存
ab -n 2000 -c 100 -p postfile -T ‘application/x-www-form-urlencoded’ http://192.168.137.1:8080/seckill/doseckill 已经秒光，可是还有库存。原因，就是乐观锁导致很多请求都失败。先点的没秒到，后点的可能秒到了。
11.5.3.连接超时，通过连接池解决
11.5.4.连接池
节省每次连接redis服务带来的消耗，把连接好的实例反复利用。 通过参数管理连接的行为 代码见项目中
链接池参数 MaxTotal：控制一个pool可分配多少个jedis实例，通过pool.getResource()来获取；如果赋值为-1，则表示不限制；如果pool已经分配了MaxTotal个jedis实例，则此时pool的状态为exhausted。 maxIdle：控制一个pool最多有多少个状态为idle(空闲)的jedis实例； MaxWaitMillis：表示当borrow一个jedis实例时，最大的等待毫秒数，如果超过等待时间，则直接抛JedisConnectionException； testOnBorrow：获得一个jedis实例的时候是否检查连接可用性（ping()）；如果为true，则得到的jedis实例均是可用的；
11.6.解决库存遗留问题
11.6.1.LUA脚本
11.6.2.LUA脚本在Redis中的优势
将复杂的或者多步的redis操作，写为一个脚本，一次提交给redis执行，减少反复连接redis的次数。提升性能。 LUA脚本是类似redis事务，有一定的原子性，不会被其他命令插队，可以完成一些redis事务性的操作。 但是注意redis的lua脚本功能，只有在Redis 2.6以上的版本才可以使用。 利用lua脚本淘汰用户，解决超卖问题。 redis 2.6版本以后，通过lua脚本解决争抢问题，实际上是redis 利用其单线程的特性，用任务队列的方式解决多任务并发问题。
11.7.Redis_事务_秒杀案例_代码
11.7.1.项目结构
11.7.2.第一版：简单版
老师点10次，正常秒杀 同学一起点试一试，秒杀也是正常的。这是因为还达不到并发的效果。 使用工具ab模拟并发测试，会出现超卖情况。查看库存会出现负数。
11.7.3.第二版：加事务-乐观锁(解决超卖),但出现遗留库存和连接超时
11.7.4.第三版：连接池解决超时问题
11.7.5.第四版：解决库存依赖问题，LUA脚本
Copy local userid=KEYS[1];
local prodid=KEYS[2];
local qtkey="sk:"..prodid..":qt";
local usersKey="sk:"..prodid.":usr';
local userExists=redis.call("sismember",usersKey,userid);
if tonumber(userExists)==1 then
return 2;
end
local num= redis.call("get" ,qtkey);
if tonumber(num)<=0 then
return 0;
else
redis.call("decr",qtkey);
redis.call("sadd",usersKey,userid);
end
return 1; 上面案例的代码
index.jsp
Copy <%@ page language="java" contentType="text/html; charset=UTF-8"
pageEncoding="UTF-8"%>
<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
<title>Insert title here</title>
</head>
<body>
<h1>iPhone 13 Pro !!! 1元秒杀！！！
</h1>
<form id="msform" action="${pageContext.request.contextPath}/doseckill" enctype="application/x-www-form-urlencoded">
<input type="hidden" id="prodid" name="prodid" value="0101">
<input type="button" id="miaosha_btn" name="seckill_btn" value="秒杀点我"/>
</form>
</body>
<script type="text/javascript" src="${pageContext.request.contextPath}/script/jquery/jquery-3.1.0.js"></script>
<script type="text/javascript">
$(function(){
$("#miaosha_btn").click(function(){
var url=$("#msform").attr("action");
$.post(url,$("#msform").serialize(),function(data){
if(data=="false"){
alert("抢光了" );
$("#miaosha_btn").attr("disabled",true);
}
} );
})
})
</script>
</html>
12345678910111213141516171819202122232425262728293031323334 SecKillServlet.java 请求的入口
Copy import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.xml.ws.soap.AddressingFeature.Responses;
/**
* 秒杀案例
*/
public class SecKillServlet extends HttpServlet {
private static final long serialVersionUID = 1L;
public SecKillServlet() {
super();
}
protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
String userid = new Random().nextInt(50000) +"" ;
String prodid =request.getParameter("prodid");
//boolean isSuccess=SecKill_redis.doSecKill(userid,prodid);
boolean isSuccess= SecKill_redisByScript.doSecKill(userid,prodid);
response.getWriter().print(isSuccess);
}
} jedis连接池的工具类
JedisPoolUtil.java
Copy package com.atguigu;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
public class JedisPoolUtil {
private static volatile JedisPool jedisPool = null;
private JedisPoolUtil() {
}
public static JedisPool getJedisPoolInstance() {
if (null == jedisPool) {
synchronized (JedisPoolUtil.class) {
if (null == jedisPool) {
JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxTotal(200); //最大连接数
poolConfig.setMaxIdle(32);
poolConfig.setMaxWaitMillis(100*1000); //最大等待时间
poolConfig.setBlockWhenExhausted(true);
poolConfig.setTestOnBorrow(true); // ping PONG
jedisPool = new JedisPool(poolConfig, "192.168.44.168", 6379, 60000 );
}
}
}
return jedisPool;
}
public static void release(JedisPool jedisPool, Jedis jedis) {
if (null != jedis) {
jedisPool.returnResource(jedis);
}
}
}
1234567891011121314151617181920212223242526272829303132333435363738 SecKill_redis.java 具体并发实现的代码
Copy package com.atguigu;
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.LoggerFactory;
import ch.qos.logback.core.rolling.helper.IntegerTokenConverter;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.ShardedJedisPool;
import redis.clients.jedis.Transaction;
/**
*
*/
public class SecKill_redis {
public static void main(String[] args) {
Jedis jedis =new Jedis("192.168.44.168",6379);
System.out.println(jedis.ping());
jedis.close();
}
//秒杀过程
public static boolean doSecKill(String uid,String prodid) throws IOException {
//1 uid和prodid非空判断
if(uid == null || prodid == null) {
return false;
}
//2 连接redis
//Jedis jedis = new Jedis("192.168.44.168",6379);
//通过连接池得到jedis对象
JedisPool jedisPoolInstance = JedisPoolUtil.getJedisPoolInstance();
Jedis jedis = jedisPoolInstance.getResource();
//3 拼接key
// 3.1 库存key
String kcKey = "sk:"+prodid+":qt";
// 3.2 秒杀成功用户key
String userKey = "sk:"+prodid+":user";
//监视库存，乐观锁
jedis.watch(kcKey);
//4 获取库存，如果库存null，秒杀还没有开始
String kc = jedis.get(kcKey);
if(kc == null) {
System.out.println("秒杀还没有开始，请等待");
jedis.close();
return false;
}
// 5 判断用户是否重复秒杀操作
if(jedis.sismember(userKey, uid)) {
System.out.println("已经秒杀成功了，不能重复秒杀");
jedis.close();
return false;
}
//6 判断如果商品数量，库存数量小于1，秒杀结束
if(Integer.parseInt(kc)<=0) {
System.out.println("秒杀已经结束了");
jedis.close();
return false;
}
//7 秒杀过程
//使用事务
Transaction multi = jedis.multi(); //开启事务，下面的语句会进入队列里面
//组队操作
multi.decr(kcKey);
multi.sadd(userKey,uid);
//执行
List<Object> results = multi.exec(); // 执行队列里面的语句，添加了乐观锁，对应的数据如果改变了，下面的语句就不能执行
if(results == null || results.size()==0) {
System.out.println("秒杀失败了....");
jedis.close();
return false;
}
//7.1 库存-1
//jedis.decr(kcKey);
//7.2 把秒杀成功用户添加清单里面
//jedis.sadd(userKey,uid);
System.out.println("秒杀成功了..");
jedis.close();
return true;
}
}
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102 使用了乐观锁和redis的事务操作
但是使用乐观锁还是会出现问题，库存遗留问题
所以使用下面的Lua语句，会解决这个问题
SecKill_redisByScript.java
Copy package com.atguigu;
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.LoggerFactory;
import ch.qos.logback.core.joran.conditional.ElseAction;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.ShardedJedisPool;
import redis.clients.jedis.Transaction;
public class SecKill_redisByScript {
private static final org.slf4j.Logger logger =LoggerFactory.getLogger(SecKill_redisByScript.class) ;
public static void main(String[] args) {
JedisPool jedispool = JedisPoolUtil.getJedisPoolInstance();
Jedis jedis=jedispool.getResource();
System.out.println(jedis.ping());
Set<HostAndPort> set=new HashSet<HostAndPort>();
// doSecKill("201","sk:0101");
}
static String secKillScript ="local userid=KEYS[1];\r\n" +
"local prodid=KEYS[2];\r\n" +
"local qtkey='sk:'..prodid..\":qt\";\r\n" +
"local usersKey='sk:'..prodid..\":usr\";\r\n" +
"local userExists=redis.call(\"sismember\",usersKey,userid);\r\n" +
"if tonumber(userExists)==1 then \r\n" +
" return 2;\r\n" +
"end\r\n" +
"local num= redis.call(\"get\" ,qtkey);\r\n" +
"if tonumber(num)<=0 then \r\n" +
" return 0;\r\n" +
"else \r\n" +
" redis.call(\"decr\",qtkey);\r\n" +
" redis.call(\"sadd\",usersKey,userid);\r\n" +
"end\r\n" +
"return 1" ;
static String secKillScript2 =
"local userExists=redis.call(\"sismember\",\"{sk}:0101:usr\",userid);\r\n" +
" return 1";
public static boolean doSecKill(String uid,String prodid) throws IOException {
JedisPool jedispool = JedisPoolUtil.getJedisPoolInstance();
Jedis jedis=jedispool.getResource();
//String sha1= .secKillScript;
String sha1= jedis.scriptLoad(secKillScript);
Object result= jedis.evalsha(sha1, 2, uid,prodid);
String reString=String.valueOf(result);
if ("0".equals( reString ) ) {
System.err.println("已抢空！！");
}else if("1".equals( reString ) ) {
System.out.println("抢购成功！！！！");
}else if("2".equals( reString ) ) {
System.err.println("该用户已抢过！！");
}else{
System.err.println("抢购异常！！");
}
jedis.close();
return true;
}
} redis执行Lua脚本，是原则性的。互相不干扰。完全解决了并发的问题。
12.Redis持久化之RDB
12.1.总体介绍
官网介绍：http://www.redis.io
12.2.RDB（Redis DataBase）
12.2.1.官网介绍
12.2.2.是什么
在指定的时间间隔内将内存中的数据集快照写入磁盘， 也就是行话讲的Snapshot快照，它恢复时是将快照文件直接读到内存里
12.2.3.备份是如何执行的
Redis会单独创建（fork）一个子进程来进行持久化，会先将数据写入到 一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。 整个过程中，主进程是不进行任何IO操作的，这就确保了极高的性能 如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。RDB的缺点是最后一次持久化后的数据可能丢失。
12.2.4.Fork
Fork的作用是复制一个与当前进程一样的进程。新进程的所有数据（变量、环境变量、程序计数器等） 数值都和原进程一致，但是是一个全新的进程，并作为原进程的子进程
在Linux程序中，fork()会产生一个和父进程完全相同的子进程，但子进程在此后多会exec系统调用，出于效率考虑，Linux中引入了“写时复制技术”
一般情况父进程和子进程会共用同一段物理内存，只有进程空间的各段的内容要发生变化时，才会将父进程的内容复制一份给子进程。
12.2.5.RDB持久化流程
12.2.6.dump.rdb文件
redis.conf文件中设置的
12.2.7.配置位置
rdb文件的保存路径，也可以修改。默认为Redis启动时命令行所在的目录下 dir “/myredis/”
12.2.8.如何触发RDB快照；保持策略
12.2.8.1.配置文件中默认的快照配置
12.2.8.2.命令save VS bgsave
save ：save时只管保存，其它不管，全部阻塞。手动保存。不建议。
bgsave：Redis会在后台异步进行快照操作， 快照同时还可以响应客户端请求。
可以通过lastsave 命令获取最后一次成功执行快照的时间
12.2.8.3.flushall命令
执行flushall命令，也会产生dump.rdb文件，但里面是空的，无意义
12.2.8.4.###SNAPSHOTTING快照###
12.2.8.5.Save
格式：save 秒钟 写操作次数 RDB是整个内存的压缩过的Snapshot，RDB的数据结构，可以配置复合的快照触发条件， 默认是1分钟内改了1万次，或5分钟内改了10次，或15分钟内改了1次。 禁用 不设置save指令，或者给save传入空字符串
12.2.8.6.stop-writes-on-bgsave-error
12.2.8.7.rdbcompression 压缩文件
12.2.8.8.rdbchecksum 检查完整性
12.2.8.9.rdb的备份
先通过config get dir 查询rdb文件的目录 将*.rdb的文件拷贝到别的地方
rdb的恢复 关闭Redis 先把备份的文件拷贝到工作目录下 cp dump2.rdb dump.rdb 启动Redis, 备份数据会直接加载
12.2.9.优势
适合大规模的数据恢复 对数据完整性和一致性要求不高更适合使用 节省磁盘空间 恢复速度快
12.2.10.劣势
Fork的时候，内存中的数据被克隆了一份，大致2倍的膨胀性需要考虑 虽然Redis在fork时使用了写时拷贝技术,但是如果数据庞大时还是比较消耗性能。 在备份周期在一定间隔时间做一次备份，所以如果Redis意外down掉的话，就会丢失最后一次快照后的所有修改。
12.2.11.如何停止
动态停止RDB：redis-cli config set save “”#save后给空值，表示禁用保存策略
12.2.12.小总结
13.Redis持久化之AOF
13.1.AOF（Append Only File）
13.1.1.是什么
以日志的形式来记录每个写操作（增量保存），将Redis执行过的所有写指令记录下来(读操作不记录)， 只许追加文件但不可以改写文件，redis启动之初会读取该文件重新构建数据，换言之，redis 重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作
13.1.2.AOF持久化流程
（1）客户端的请求写命令会被append追加到AOF缓冲区内； （2）AOF缓冲区根据AOF持久化策略[always,everysec,no]将操作sync同步到磁盘的AOF文件中； （3）AOF文件大小超过重写策略或手动重写时，会对AOF文件rewrite重写，压缩AOF文件容量； （4）Redis服务重启时，会重新load加载AOF文件中的写操作达到数据恢复的目的；
13.1.3.AOF默认不开启
可以在redis.conf中配置文件名称，默认为 appendonly.aof AOF文件的保存路径，同RDB的路径一致。
13.1.4.AOF和RDB同时开启，redis听谁的？
AOF和RDB同时开启，系统默认取AOF的数据（数据不会存在丢失）
13.1.5.AOF启动/修复/恢复
AOF的备份机制和性能虽然和RDB不同, 但是备份和恢复的操作同RDB一样，都是拷贝备份文件，需要恢复时再拷贝到Redis工作目录下，启动系统即加载。
正常恢复 修改默认的appendonly no，改为yes 将有数据的aof文件复制一份保存到对应目录(查看目录：config get dir) 恢复：重启redis然后重新加载
异常恢复 修改默认的appendonly no，改为yes 如遇到AOF文件损坏，通过/usr/local/bin/redis-check-aof–fix appendonly.aof进行恢复 备份被写坏的AOF文件 恢复：重启redis，然后重新加载
13.1.6.AOF同步频率设置
appendfsync always 始终同步，每次Redis的写入都会立刻记入日志；性能较差但数据完整性比较好
appendfsync everysec 每秒同步，每秒记入日志一次，如果宕机，本秒的数据可能丢失。
appendfsync no redis不主动进行同步，把同步时机交给操作系统。
13.1.7.Rewrite压缩
1是什么： AOF采用文件追加方式，文件会越来越大为避免出现此种情况，新增了重写机制, 当AOF文件的大小超过所设定的阈值时，Redis就会启动AOF文件的内容压缩， 只保留可以恢复数据的最小指令集.可以使用命令bgrewriteaof
2重写原理，如何实现重写 AOF文件持续增长而过大时，会fork出一条新进程来将文件重写(也是先写临时文件最后再rename)，redis4.0版本后的重写，是指上就是把rdb 的快照，以二级制的形式附在新的aof头部，作为已有的历史数据，替换掉原来的流水账操作。 no-appendfsync-on-rewrite：
如果 no-appendfsync-on-rewrite=yes ,不写入aof文件只写入缓存，用户请求不会阻塞，但是在这段时间如果宕机会丢失这段时间的缓存数据。（降低数据安全性，提高性能） 如果 no-appendfsync-on-rewrite=no, 还是会把数据往磁盘里刷，但是遇到重写操作，可能会发生阻塞。（数据安全，但是性能降低） 触发机制，何时重写 Redis会记录上次重写时的AOF大小，默认配置是当AOF文件大小是上次rewrite后大小的一倍且文件大于64M时触发 重写虽然可以节约大量磁盘空间，减少恢复时间。但是每次重写还是有一定的负担的，因此设定Redis要满足一定条件才会进行重写。 auto-aof-rewrite-percentage：设置重写的基准值，文件达到100%时开始重写（文件是原来重写后文件的2倍时触发） auto-aof-rewrite-min-size：设置重写的基准值，最小文件64MB。达到这个值开始重写。 例如：文件达到70MB开始重写，降到50MB，下次什么时候开始重写？100MB 系统载入时或者上次重写完毕时，Redis会记录此时AOF大小，设为base_size, 如果Redis的AOF当前大小>= base_size +base_size*100% (默认)且当前大小>=64mb(默认)的情况下，Redis会对AOF进行重写。
3、重写流程
（1）bgrewriteaof触发重写，判断是否当前有bgsave或bgrewriteaof在运行，如果有，则等待该命令结束后再继续执行。 （2）主进程fork出子进程执行重写操作，保证主进程不会阻塞。 （3）子进程遍历redis内存中数据到临时文件，客户端的写请求同时写入aof_buf缓冲区和aof_rewrite_buf重写缓冲区保证原AOF文件完整以及新AOF文件生成期间的新的数据修改动作不会丢失。 （4）1).子进程写完新的AOF文件后，向主进程发信号，父进程更新统计信息。2).主进程把aof_rewrite_buf中的数据写入到新的AOF文件。 （5）使用新的AOF文件覆盖旧的AOF文件，完成AOF重写。
13.1.8.优势
13.1.9.劣势
比起RDB占用更多的磁盘空间。 恢复备份速度要慢。 每次读写都同步的话，有一定的性能压力。 存在个别Bug，造成恢复不能。
13.1.10. 小总结
13.2.总结(Which one)
13.2.1.用哪个好
官方推荐两个都启用。 如果对数据不敏感，可以选单独用RDB。 不建议单独用 AOF，因为可能会出现Bug。 如果只是做纯内存缓存，可以都不用。
13.2.2.官网建议
Copy 因为RDB文件只用作后备用途，建议只在Slave上持久化RDB文件，而且只要15分钟备份一次就够了，只保留save 900 1这条规则。
如果使用AOF，好处是在最恶劣情况下也只会丢失不超过两秒数据，启动脚本较简单只load自己的AOF文件就可以了。
代价,一是带来了持续的IO，二是AOF rewrite的最后将rewrite过程中产生的新数据写到新文件造成的阻塞几乎是不可避免的。
只要硬盘许可，应该尽量减少AOF rewrite的频率，AOF重写的基础大小默认值64M太小了，可以设到5G以上。
默认超过原大小100%大小时重写可以改到适当的数值。
12345678910 14.Redis_主从复制
14.1.是什么
主机数据更新后根据配置和策略， 自动同步到备机的master/slaver机制，Master以写为主，Slave以读为主
14.2.能干嘛
读写分离，性能扩展 容灾快速恢复
14.3.怎么玩：主从复制
拷贝多个redis.conf文件include(写绝对路径) 开启daemonize yes Pid文件名字pidfile 指定端口port Log文件名字 dump.rdb名字dbfilename Appendonly 关掉或者换名字
14.3.1.新建redis6379.conf，填写以下内容
include /myredis/redis.conf pidfile /var/run/redis_6379.pid port 6379 dbfilename dump6379.rdb
14.3.2.新建redis6380.conf，填写以下内容
14.3.3.新建redis6381.conf，填写以下内容
slave-priority 10 设置从机的优先级，值越小，优先级越高，用于选举主机时使用。默认100
14.3.4.启动三台redis服务器
14.3.5.查看系统进程，看看三台服务器是否启动
14.3.6.查看三台主机运行情况
info replication 打印主从复制的相关信息
14.3.7.配从(库)不配主(库)
slaveof 成为某个实例的从服务器 1、在6380和6381上执行: slaveof 127.0.0.1 6379
4、从机重启需重设：slaveof 127.0.0.1 6379
14.4.常用3招
14.4.1.一主二仆
切入点问题？slave1、slave2是从头开始复制还是从切入点开始复制?比如从k4进来，那之前的k1,k2,k3是否也可以复制？ 从机是否可以写？set可否？ 主机shutdown后情况如何？从机是上位还是原地待命？ 主机又回来了后，主机新增记录，从机还能否顺利复制？ 其中一台从机down后情况如何？依照原有它能跟上大部队吗？
14.4.2.薪火相传
上一个Slave可以是下一个slave的Master，Slave同样可以接收其他 slaves的连接和同步请求，那么该slave作为了链条中下一个的master, 可以有效减轻master的写压力,去中心化降低风险。 用 slaveof 中途变更转向:会清除之前的数据，重新建立拷贝最新的 风险是一旦某个slave宕机，后面的slave都没法备份 主机挂了，从机还是从机，无法写数据了
14.4.3.反客为主
14.5.复制原理
Slave启动成功连接到master后会发送一个sync命令
Master接到命令启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令， 在后台进程执行完毕之后，master将传送整个数据文件到slave,以完成一次完全同步
全量复制：而slave服务在接收到数据库文件数据后，将其存盘并加载到内存中。
增量复制：Master继续将新的所有收集到的修改命令依次传给slave,完成同步
但是只要是重新连接master,一次完全同步（全量复制)将被自动执行
14.6.哨兵模式(sentinel)
14.6.1.是什么
反客为主的自动版，能够后台监控主机是否故障，如果故障了根据投票数自动将从库转换为主库
14.6.2.怎么玩(使用步骤)
14.6.2.1.调整为一主二仆模式，6379带着6380、6381
14.6.2.2.自定义的/myredis目录下新建sentinel.conf文件，名字绝不能错
14.6.2.3.配置哨兵,填写内容
sentinel monitor mymaster 127.0.0.1 6379 1 其中mymaster为监控对象起的服务器名称， 1 为至少有多少个哨兵同意迁移的数量。
14.6.2.4.启动哨兵
/usr/local/bin redis做压测可以用自带的redis-benchmark工具 执行redis-sentinel /myredis/sentinel.conf
14.6.2.5.当主机挂掉，从机选举中产生新的主机
(大概10秒左右可以看到哨兵窗口日志，切换了新的主机) 哪个从机会被选举为主机呢？根据优先级别：slave-priority 原主机重启后会变为从机。
14.6.2.6.复制延时
由于所有的写操作都是先在Master上操作，然后同步更新到Slave上，所以从Master同步到Slave机器有一定的延迟，当系统很繁忙的时候，延迟问题会更加严重，Slave机器数量的增加也会使这个问题更加严重。
14.6.3.故障恢复
14.6.4.主从复制
Copy private static JedisSentinelPool jedisSentinelPool=null;
public static Jedis getJedisFromSentinel(){
if(jedisSentinelPool==null){
Set<String> sentinelSet=new HashSet<>();
sentinelSet.add("192.168.11.103:26379"); //主机的地址，端口号
JedisPoolConfig jedisPoolConfig =new JedisPoolConfig();
jedisPoolConfig.setMaxTotal(10); //最大可用连接数
jedisPoolConfig.setMaxIdle(5); //最大闲置连接数
jedisPoolConfig.setMinIdle(5); //最小闲置连接数
jedisPoolConfig.setBlockWhenExhausted(true); //连接耗尽是否等待
jedisPoolConfig.setMaxWaitMillis(2000); //等待时间
jedisPoolConfig.setTestOnBorrow(true); //取连接的时候进行一下测试 ping pong
jedisSentinelPool=new JedisSentinelPool("mymaster",sentinelSet,jedisPoolConfig);
return jedisSentinelPool.getResource();
}else{
return jedisSentinelPool.getResource();
}
}
1234567891011121314151617181920212223 15.Redis集群
15.1.问题
容量不够，redis如何进行扩容？ 并发写操作， redis如何分摊？ 另外，主从模式，薪火相传模式，主机宕机，导致ip地址发生变化，应用程序中配置需要修改对应的主机地址、端口等信息。 之前通过代理主机来解决，但是redis3.0中提供了解决方案。就是无中心化集群配置。
15.2.什么是集群
Redis 集群实现了对Redis的水平扩容，即启动N个redis节点，将整个数据库分布存储在这N个节点中，每个节点存储总数据的1/N。 Redis 集群通过分区（partition）来提供一定程度的可用性（availability）： 即使集群中有一部分节点失效或者无法进行通讯， 集群也可以继续处理命令请求。
15.3.删除持久化数据
将rdb,aof文件都删除掉。
15.4.制作6个实例，6379,6380,6381,6389,6390,6391
15.4.1.配置基本信息
开启daemonize yes Pid文件名字 指定端口 Log文件名字 Dump.rdb名字 Appendonly 关掉或者换名字
15.4.2.redis cluster配置修改
cluster-enabled yes 打开集群模式 cluster-config-file nodes-6379.conf 设定节点配置文件名 cluster-node-timeout 15000 设定节点失联时间，超过该时间（毫秒），集群自动进行主从切换。
Copy include /home/bigdata/redis.conf
port 6379
pidfile "/var/run/redis_6379.pid"
dbfilename "dump6379.rdb"
dir "/home/bigdata/redis_cluster"
logfile "/home/bigdata/redis_cluster/redis_err_6379.log"
cluster-enabled yes
cluster-config-file nodes-6379.conf
cluster-node-timeout 15000
123456789 15.4.3.修改好redis6379.conf文件，拷贝多个redis.conf文件
15.4.4.使用查找替换修改另外5个文件
例如：:%s/6379/6380
15.4.5.启动6个redis服务
15.5.将六个节点合成一个集群
组合之前，请确保所有redis实例启动后，nodes-xxxx.conf文件都生成正常。
合体： cd /opt/redis-6.2.1/src
Copy redis-cli --cluster create --cluster-replicas 1 192.168.11.101:6379 192.168.11.101:6380 192.168.11.101:6381
192.168.11.101:6389 192.168.11.101:6390 192.168.11.101:6391
12 此处不要用127.0.0.1， 请用真实IP地址 –replicas 1 采用最简单的方式配置集群，一台主机，一台从机，正好三组。
普通方式登录 可能直接进入读主机，存储数据时，会出现MOVED重定向操作。所以，应该以集群方式登录。
15.6.-c 采用集群策略连接，设置数据会自动切换到相应的写主机
15.7.通过 cluster nodes 命令查看集群信息
15.8.redis cluster 如何分配这六个节点?
一个集群至少要有三个主节点。 选项 --cluster-replicas 1 表示我们希望为集群中的每个主节点创建一个从节点。 分配原则尽量保证每个主数据库运行在不同的IP地址，每个从库和主库不在一个IP地址上。
15.9.什么是slots
15.10.在集群中录入值
在redis-cli每次录入、查询键值，redis都会计算出该key应该送往的插槽，如果不是该客户端对应服务器的插槽，redis会报错，并告知应前往的redis实例地址和端口。 redis-cli客户端提供了 –c 参数实现自动重定向。 如 redis-cli -c –p 6379 登入后，再录入、查询键值对可以自动重定向。
15.12.查询集群中的值
15.13.故障恢复
15.14.集群的Jedis开发
即使连接的不是主机，集群会自动切换主机存储。主机写，从机读。 无中心化主从集群。无论从哪台主机写的数据，其他主机上都能读到数据。
Copy public class JedisClusterTest {
public static void main(String[] args) {
Set<HostAndPort>set =new HashSet<HostAndPort>();
set.add(new HostAndPort("192.168.31.211",6379));
JedisCluster jedisCluster=new JedisCluster(set);
jedisCluster.set("k1", "v1");
System.out.println(jedisCluster.get("k1"));
}
}
123456789 15.15.Redis 集群提供了以下好处
实现扩容 分摊压力 无中心配置相对简单
15.16.Redis 集群的不足
多键操作是不被支持的 多键的Redis事务是不被支持的。lua脚本不被支持 由于集群方案出现较晚，很多公司已经采用了其他的集群方案，而代理或者客户端分片的方案想要迁移至redis cluster，需要整体迁移而不是逐步过渡，复杂度较大。
16.Redis应用问题解决
16.1.缓存穿透
16.1.1.问题描述
key对应的数据在数据源并不存在，每次针对此key的请求从缓存获取不到，请求都会压到数据源，从而可能压垮数据源。比如用一个不存在的用户id获取用户信息，不论缓存还是数据库都没有，若黑客利用此漏洞进行攻击可能压垮数据库。
16.1.2.解决方案
一个一定不存在缓存及查询不到的数据，由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义。
解决方案：
对空值缓存：如果一个查询返回的数据为空（不管是数据是否不存在），我们仍然把这个空结果（null）进行缓存，设置空结果的过期时间会很短，最长不超过五分钟
设置可访问的名单（白名单）： 使用bitmaps类型定义一个可以访问的名单，名单id作为bitmaps的偏移量，每次访问和bitmap里面的id进行比较，如果访问id不在bitmaps里面，进行拦截，不允许访问。 （
采用布隆过滤器：(布隆过滤器（Bloom Filter）是1970年由布隆提出的。它实际上是一个很长的二进制向量(位图)和一系列随机映射函数（哈希函数）。布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都远远超过一般的算法，缺点是有一定的误识别率和删除困难。) 将所有可能存在的数据哈希到一个足够大的bitmaps中，一个一定不存在的数据会被 这个bitmaps拦截掉，从而避免了对底层存储系统的查询压力。
进行实时监控：当发现Redis的命中率开始急速降低，需要排查访问对象和访问的数据，和运维人员配合，可以设置黑名单限制服务
16.2.缓存击穿
16.2.1.问题描述
16.2.2.解决方案
key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题：缓存被“击穿”的问题。
解决问题：
预先设置热门数据：在redis高峰访问之前，把一些热门数据提前存入到redis里面，加大这些热门数据key的时长
实时调整：现场监控哪些数据热门，实时调整key的过期时长
使用锁：
就是在缓存失效的时候（判断拿出来的值为空），不是立即去load db。
先使用缓存工具的某些带成功操作返回值的操作（比如Redis的SETNX） 去set一个mutex key
当操作返回成功时，再进行load db的操作，并回设缓存,最后删除mutex key；
当操作返回失败，证明有线程在load db，当前线程睡眠一段时间再重试整个get缓存的方法。
16.3.缓存雪崩
16.3.1.问题描述
key对应的数据存在，但在redis中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。 缓存雪崩与缓存击穿的区别在于这里针对很多key缓存，前者则是某一个key
正常访问
16.3.2.解决方案
缓存失效时的雪崩效应对底层系统的冲击非常可怕！
解决方案：
构建多级缓存架构：nginx缓存 + redis缓存 +其他缓存（ehcache等）
使用锁或队列： 用加锁或者队列的方式保证来保证不会有大量的线程对数据库一次性进行读写，从而避免失效时大量的并发请求落到底层存储系统上。不适用高并发情况
设置过期标志更新缓存： 记录缓存数据是否过期（设置提前量），如果过期会触发通知另外的线程在后台去更新实际key的缓存。
将缓存失效时间分散开： 比如我们可以在原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这样
16.4.分布式锁
16.4.1.问题描述
随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！
分布式锁主流的实现方案：
每一种分布式锁解决方案都有各自的优缺点：
可靠性：zookeeper最高 这里，我们就基于redis实现分布式锁。
16.4.2.解决方案：使用redis实现分布式锁
redis:命令
Copy # set sku:1:info “OK” NX PX 10000
1 EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。 PX millisecond ：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecond value 。 NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。 XX ：只在键已经存在时，才对键进行设置操作。
获取成功，执行业务逻辑{从db获取数据，放入缓存}，执行完成释放锁（del）
16.4.3.编写代码
Redis: set num 0
Copy @GetMapping("testLock")
public void testLock(){
//1获取锁，setne
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111");
//2获取锁成功、查询num的值
if(lock){
Object value = redisTemplate.opsForValue().get("num");
//2.1判断num为空return
if(StringUtils.isEmpty(value)){
return;
}
//2.2有值就转成成int
int num = Integer.parseInt(value+"");
//2.3把redis的num加1
redisTemplate.opsForValue().set("num", ++num);
//2.4释放锁，del
redisTemplate.delete("lock");
}else{
//3获取锁失败、每隔0.1秒再获取
try {
Thread.sleep(100);
testLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
12345678910111213141516171819202122232425262728 基本实现。 问题：setnx刚好获取到锁，业务逻辑出现异常，导致锁无法释放 解决：设置过期时间，自动释放锁。
16.4.4.优化之设置锁的过期时间
设置过期时间有两种方式：
首先想到通过expire设置过期时间（缺乏原子性：如果在setnx和expire之间出现异常，锁也无法释放）
场景：如果业务逻辑的执行时间是7s。执行流程如下 1.index1业务逻辑没执行完，3秒后锁被自动释放。 2.index2获取到锁，执行业务逻辑，3秒后锁被自动释放。 3.index3获取到锁，执行业务逻辑 4.index1业务逻辑执行完成，开始调用del释放锁，这时释放的是index3的锁，导致index3的业务只执行1s就被别人释放。 最终等于没锁的情况。
解决：setnx获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这个值，判断是否自己的锁
16.4.5.优化之UUID防误删
16.4.6.优化之LUA脚本保证删除的原子性
Copy @GetMapping("testLockLua")
public void testLockLua() {
//1 声明一个uuid ,将做为一个value 放入我们的key所对应的值中
String uuid = UUID.randomUUID().toString();
//2 定义一个锁：lua 脚本可以使用同一把锁，来实现删除！
String skuId = "25"; // 访问skuId 为25号的商品 100008348542
String locKey = "lock:" + skuId; // 锁住的是每个商品的数据
// 3 获取锁
Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 3, TimeUnit.SECONDS);
// 第一种： lock 与过期时间中间不写任何的代码。
// redisTemplate.expire("lock",10, TimeUnit.SECONDS);//设置过期时间
// 如果true
if (lock) {
// 执行的业务逻辑开始
// 获取缓存中的num 数据
Object value = redisTemplate.opsForValue().get("num");
// 如果是空直接返回
if (StringUtils.isEmpty(value)) {
return;
}
// 不是空 如果说在这出现了异常！ 那么delete 就删除失败！ 也就是说锁永远存在！
int num = Integer.parseInt(value + "");
// 使num 每次+1 放入缓存
redisTemplate.opsForValue().set("num", String.valueOf(++num));
/*使用lua脚本来锁*/
// 定义lua 脚本
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
// 使用redis执行lua执行
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
redisScript.setScriptText(script);
// 设置一下返回值类型 为Long
// 因为删除判断的时候，返回的0,给其封装为数据类型。如果不封装那么默认返回String 类型，
// 那么返回字符串与0 会有发生错误。
redisScript.setResultType(Long.class);
// 第一个要是script 脚本 ，第二个需要判断的key，第三个就是key所对应的值。
redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid);
} else {
// 其他线程等待
try {
// 睡眠
Thread.sleep(1000);
// 睡醒了之后，调用方法。
testLockLua();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849 Lua 脚本详解：
1.定义key，key应该是为每个sku定义的，也就是每个sku有一把锁。 String locKey =“lock:”+skuId; // 锁住的是每个商品的数据 Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid,3,TimeUnit.SECONDS);
16.4.7.总结
Copy // 2. 释放锁 del
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
// 设置lua脚本返回的数据类型
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
// 设置lua脚本返回类型为Long
redisScript.setResultType(Long.class);
redisScript.setScriptText(script);
redisTemplate.execute(redisScript, Arrays.asList("lock"),uuid);
12345678 3、重试
Copy Thread.sleep(500);
testLock();
12 为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：
不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。
