文章目录
- 1 Redis事务
- 1.1 Redis事务简介
- 1.2 Redis事务操作过程
- 1.2.1 开启事务--->执行事务
- 1.2.2 取消事务(discurd)
- 1.3 事务错误
- 1.3.1 编译期异常
- 1.3.2 运行时异常
- 2 Redis实现乐观锁
- 2.1 乐观锁和悲观锁
- 2.2 正常执行
- 2.3 测试异常执行
- 3 Jedis
- 3.1 导入依赖
- 3.2 编码测试
- 3.2.1 修改配置文件
- 3.3 常用API
- 4 Jedis实现事务操作
- 5 SpringBoot集成Jedis
- 5.1 新建springboot项目
- 5.2 源码分析
- 5.3 配置redis
- 5.4 测试一下
- 5.5 序列化解决中文乱码问题
- 5.6 Jedis工具类
- 6 Redis配置文件详解
- 6.1 容量单位
- 6.2 使用多个配置
- 6.3 网络配置
- 6.4 通用GENERAL
- 6.5日志输出级别
- 6.6 日志输出文件
- 6.7 持久化规则(快照)
- 6.8 RDB文件相关
- 6.9 安全
- 6.10 限制
- 6.11 APPEND ONLY 模式 aof配置
- 7 redis持久化
- 7.1 RDB
- 7.1.1 什么是RDB
- 7.1.2 工作原理
- 7.1.3 触发机制
- 7.1.4 触发持久化规则
- 7.2 AOF
- 7.2.1 什么是AOF
- 7.2.2 aof优缺点
- 7.3 RDB和AOF的对比和选择
- 如何选择使用哪种持久化方式？
- 8 Redis发布和订阅
- 8.1 相关命令
- 8.2 实例
- 8.3 原理
- 8.4 缺点及应用
- 9 主从复制
- 9.1 概念
- 9.2 作用
- 9.3 为什么使用集群
- 9.4 环境配置
- 9.5 一主二从配置
- 9.6 使用规则
- 10 哨兵模式
- 10.1 哨兵的作用：
- 10.2 哨兵的核心配置
- 10.3测试
- 10.4 哨兵模式优缺点
- 10.5 哨兵模式的全部配置
- 11 缓存穿透和雪崩
- 11.1 缓存穿透
- 11.2 缓存击穿
- 11.3 缓存雪崩
- 12 学习链接
1 Redis事务
1.1 Redis事务简介
Redis的单条命令是保证原子性的，但是redis事务不能保证原子性。
Redis事务本质：一组命令的集合。
----------------- 队列 set set set 执行 -------------------
事务中每条命令都会被序列化，执行过程中按顺序执行，不允许其他命令进行干扰。
- 一次性
- 顺序性
- 排他性
① Redis事务没有隔离级别的概念
② Redis单条命令是保证原子性的，但是事务不保证原子性！
1.2 Redis事务操作过程
1.2.1 开启事务—>执行事务
- 开启事务（
multi） - 命令入队
- 执行事务（
exec）
所有事务中的命令在加入时都没有被执行，直到提交时才会开始执行(Exec)一次性完成。
127.0.0.1:6379> multi # 开启事务
OK
127.0.0.1:6379> set k1 v1 # 命令入队
QUEUED
127.0.0.1:6379> set k2 v2 # ..
QUEUED
127.0.0.1:6379> get k1
QUEUED
127.0.0.1:6379> set k3 v3
QUEUED
127.0.0.1:6379> keys *
QUEUED
127.0.0.1:6379> exec # 事务执行
1) OK
2) OK
3) "v1"
4) OK
5) 1) "k3"
2) "k2"
3) "k1"
1.2.2 取消事务(discurd)
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> DISCARD # 放弃事务
OK
127.0.0.1:6379> EXEC
(error) ERR EXEC without MULTI # 当前未开启事务
127.0.0.1:6379> get k1 # 被放弃事务中命令并未执行
(nil)
1.3 事务错误
1.3.1 编译期异常
代码语法错误（编译时异常）所有的命令都不执行
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> error k1 # 这是一条语法错误命令
(error) ERR unknown command `error`, with args beginning with: `k1`, # 会报错但是不影响后续命令入队
127.0.0.1:6379> get k2
QUEUED
127.0.0.1:6379> EXEC
(error) EXECABORT Transaction discarded because of previous errors. # 执行报错
127.0.0.1:6379> get k1
(nil) # 其他命令并没有被执行
1.3.2 运行时异常
代码逻辑错误 (运行时异常) **其他命令可以正常执行 ** >>> 所以不保证事务原子性
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> INCR k1 # 这条命令逻辑错误（对字符串进行增量）
QUEUED
127.0.0.1:6379> get k2
QUEUED
127.0.0.1:6379> exec
1) OK
2) OK
3) (error) ERR value is not an integer or out of range # 运行时报错
4) "v2" # 其他命令正常执行
# 虽然中间有一条命令报错了，但是后面的指令依旧正常执行成功了。
# 所以说Redis单条指令保证原子性，但是Redis事务不能保证原子性。
2 Redis实现乐观锁
2.1 乐观锁和悲观锁
悲观锁：
- 很悲观，认为什么时候都会出现问题，无论做什么都会加锁
乐观锁：
- 很乐观，认为什么时候都不会出现问题，所以不会上锁！更新数据的时候去判断一下，在此期间是否有人修改过这个数据
- 获取version
- 更新的时候比较version
使用watch key监控指定数据，相当于乐观锁加锁。
2.2 正常执行
127.0.0.1:6379> set money 100 # 设置余额:100
OK
127.0.0.1:6379> set use 0 # 支出使用:0
OK
127.0.0.1:6379> watch money # 监视money (上锁)
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> DECRBY money 20
QUEUED
127.0.0.1:6379> INCRBY use 20
QUEUED
127.0.0.1:6379> exec # 监视值没有被中途修改，事务正常执行
1) (integer) 80
2) (integer) 20
2.3 测试异常执行
测试多线程修改值，使用watch可以当做redis的乐观锁操作（相当于getversion）
我们启动另外一个客户端模拟插队线程。
线程1：
127.0.0.1:6379> watch money # money上锁
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> DECRBY money 20
QUEUED
127.0.0.1:6379> INCRBY use 20
QUEUED
127.0.0.1:6379> # 此时事务并没有执行
模拟线程插队，线程2：
127.0.0.1:6379> INCRBY money 500 # 修改了线程一中监视的money
(integer) 600
回到线程1，执行事务：
127.0.0.1:6379> EXEC # 执行之前，另一个线程修改了我们的值，这个时候就会导致事务执行失败
(nil) # 没有结果，说明事务执行失败
127.0.0.1:6379> get money # 线程2 修改生效
"600"
127.0.0.1:6379> get use # 线程1事务执行失败，数值没有被修改
"0"
解锁获取最新值，然后再加锁进行事务。
unwatch进行解锁。
== 注意：每次提交执行exec后都会自动释放锁，不管是否成功 ==
3 Jedis
3.1 导入依赖
<dependencies>
<!--导入jredis的包-->
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.3.0</version>
</dependency>
<!--fastjson-->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.75</version>
</dependency>
</dependencies>
3.2 编码测试
3.2.1 修改配置文件
(不修改可能会报错，修改一次一劳永逸)
注释掉bind，protected-mode为no
查看远程服务器的防火墙状态
systemctl status firewalld
如果为开启状态，执行以下代码：
#查看防火墙状态
firewall-cmd --state
#开放6379端口
firewall-cmd --permanent --add-port=6379/tcp
#查看端口，执行
firewall-cmd --permanent --query-port=6379/tcp
firewall-cmd --permanent --list-ports
#重启防火墙
firewall-cmd --reload
执行完上面代码或者防火墙未开启，继续执行下面代码：
redis-cli -p 6379 shutdown
redis-server redis_config/redis.conf
redis-cli -p 6379
测试一下是否可以连接成功:
public class TestPing {
public static void main(String[] args) {
Jedis jedis = new Jedis("120.79.99.207",6379);
System.out.println(jedis.ping());
}
}
3.3 常用API
（需要时可以百度查,和原生redisAPI非常相似）
4 Jedis实现事务操作
public class TestTX {
public static void main(String[] args) {
Jedis jedis = new Jedis("120.79.99.207", 6379);
jedis.flushDB();
JSONObject jsonObject = new JSONObject();
jsonObject.put("hello", "world");
jsonObject.put("name", "wskh");
// 开启事务
Transaction multi = jedis.multi();
String result = jsonObject.toJSONString();
// 乐观锁
// jedis.watch(result);
try {
multi.set("user1", result);
multi.set("user2", result);
// 事务成功，执行事务
multi.exec();
}catch (Exception e){
// 事务失败，放弃事务
multi.discard();
} finally {
// 最后，关闭连接
System.out.println(jedis.get("user1"));
System.out.println(jedis.get("user2"));
jedis.close();
}
}
}
5 SpringBoot集成Jedis
5.1 新建springboot项目
5.2 源码分析
springboot 2.x后 ，原来使用的 Jedis 被 lettuce 替换。
5.3 配置redis
# 配置redis
spring.redis.host= 120.79.99.207
spring.redis.port= 6379
5.4 测试一下
@SpringBootTest
class RedisSpringbootApplicationTests {
@Autowired
private RedisTemplate redisTemplate;
@Test
void contextLoads() {
// redisTemplate 操作不同的数据类型，api和我们的指令是一样的
// opsForValue 操作字符串 类似String
// opsForList 操作List 类似List
// opsForHash
// opsForZSet
// opsForGeo
// opsForHyperLogLog
// 除了基本的操作，我们常用的方法都可以直接通过redisTemplate操作，比如事务和基本的CRUD
// 获取连接对象
//RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
//connection.flushDb();
//connection.flushAll();
redisTemplate.opsForValue().set("mykey","王孙康宏");
System.out.println(redisTemplate.opsForValue().get("mykey"));
}
}
测试结果：
成功~~~
5.5 序列化解决中文乱码问题
如果直接用jedis提供的api进行存储是不能存储中文的，会出现乱码问题。所以我们需要通过序列化的手段解决中文乱码问题
5.6 Jedis工具类
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
* Redis工具类*/
public class RedisUtil {
private StringRedisTemplate redisTemplate;
public void setRedisTemplate(StringRedisTemplate redisTemplate) {
this.redisTemplate = redisTemplate;
}
public StringRedisTemplate getRedisTemplate() {
return this.redisTemplate;
}
/** -------------------key相关操作--------------------- */
/**
* 删除key
*
* @param key
*/
public void delete(String key) {
redisTemplate.delete(key);
}
/**
* 批量删除key
*
* @param keys
*/
public void delete(Collection<String> keys) {
redisTemplate.delete(keys);
}
/**
* 序列化key
*
* @param key
* @return
*/
public byte[] dump(String key) {
return redisTemplate.dump(key);
}
/**
* 是否存在key
*
* @param key
* @return
*/
public Boolean hasKey(String key) {
return redisTemplate.hasKey(key);
}
/**
* 设置过期时间
*
* @param key
* @param timeout
* @param unit
* @return
*/
public Boolean expire(String key, long timeout, TimeUnit unit) {
return redisTemplate.expire(key, timeout, unit);
}
/**
* 设置过期时间
*
* @param key
* @param date
* @return
*/
public Boolean expireAt(String key, Date date) {
return redisTemplate.expireAt(key, date);
}
/**
* 查找匹配的key
*
* @param pattern
* @return
*/
public Set<String> keys(String pattern) {
return redisTemplate.keys(pattern);
}
/**
* 将当前数据库的 key 移动到给定的数据库 db 当中
*
* @param key
* @param dbIndex
* @return
*/
public Boolean move(String key, int dbIndex) {
return redisTemplate.move(key, dbIndex);
}
/**
* 移除 key 的过期时间，key 将持久保持
*
* @param key
* @return
*/
public Boolean persist(String key) {
return redisTemplate.persist(key);
}
/**
* 返回 key 的剩余的过期时间
*
* @param key
* @param unit
* @return
*/
public Long getExpire(String key, TimeUnit unit) {
return redisTemplate.getExpire(key, unit);
}
/**
* 返回 key 的剩余的过期时间
*
* @param key
* @return
*/
public Long getExpire(String key) {
return redisTemplate.getExpire(key);
}
/**
* 从当前数据库中随机返回一个 key
*
* @return
*/
public String randomKey() {
return redisTemplate.randomKey();
}
/**
* 修改 key 的名称
*
* @param oldKey
* @param newKey
*/
public void rename(String oldKey, String newKey) {
redisTemplate.rename(oldKey, newKey);
}
/**
* 仅当 newkey 不存在时，将 oldKey 改名为 newkey
*
* @param oldKey
* @param newKey
* @return
*/
public Boolean renameIfAbsent(String oldKey, String newKey) {
return redisTemplate.renameIfAbsent(oldKey, newKey);
}
/**
* 返回 key 所储存的值的类型
*
* @param key
* @return
*/
public DataType type(String key) {
return redisTemplate.type(key);
}
/** -------------------string相关操作--------------------- */
/**
* 设置指定 key 的值
* @param key
* @param value
*/
public void set(String key, String value) {
redisTemplate.opsForValue().set(key, value);
}
/**
* 获取指定 key 的值
* @param key
* @return
*/
public String get(String key) {
return redisTemplate.opsForValue().get(key);
}
/**
* 返回 key 中字符串值的子字符
* @param key
* @param start
* @param end
* @return
*/
public String getRange(String key, long start, long end) {
return redisTemplate.opsForValue().get(key, start, end);
}
/**
* 将给定 key 的值设为 value ，并返回 key 的旧值(old value)
*
* @param key
* @param value
* @return
*/
public String getAndSet(String key, String value) {
return redisTemplate.opsForValue().getAndSet(key, value);
}
/**
* 对 key 所储存的字符串值，获取指定偏移量上的位(bit)
*
* @param key
* @param offset
* @return
*/
public Boolean getBit(String key, long offset) {
return redisTemplate.opsForValue().getBit(key, offset);
}
/**
* 批量获取
*
* @param keys
* @return
*/
public List<String> multiGet(Collection<String> keys) {
return redisTemplate.opsForValue().multiGet(keys);
}
/**
* 设置ASCII码, 字符串'a'的ASCII码是97, 转为二进制是'01100001', 此方法是将二进制第offset位值变为value
*
* @param key 位置
* @param value
* 值,true为1, false为0
* @return
*/
public boolean setBit(String key, long offset, boolean value) {
return redisTemplate.opsForValue().setBit(key, offset, value);
}
/**
* 将值 value 关联到 key ，并将 key 的过期时间设为 timeout
*
* @param key
* @param value
* @param timeout
* 过期时间
* @param unit
* 时间单位, 天:TimeUnit.DAYS 小时:TimeUnit.HOURS 分钟:TimeUnit.MINUTES
* 秒:TimeUnit.SECONDS 毫秒:TimeUnit.MILLISECONDS
*/
public void setEx(String key, String value, long timeout, TimeUnit unit) {
redisTemplate.opsForValue().set(key, value, timeout, unit);
}
/**
* 只有在 key 不存在时设置 key 的值
*
* @param key
* @param value
* @return 之前已经存在返回false,不存在返回true
*/
public boolean setIfAbsent(String key, String value) {
return redisTemplate.opsForValue().setIfAbsent(key, value);
}
/**
* 用 value 参数覆写给定 key 所储存的字符串值，从偏移量 offset 开始
*
* @param key
* @param value
* @param offset
* 从指定位置开始覆写
*/
public void setRange(String key, String value, long offset) {
redisTemplate.opsForValue().set(key, value, offset);
}
/**
* 获取字符串的长度
*
* @param key
* @return
*/
public Long size(String key) {
return redisTemplate.opsForValue().size(key);
}
/**
* 批量添加
*
* @param maps
*/
public void multiSet(Map<String, String> maps) {
redisTemplate.opsForValue().multiSet(maps);
}
/**
* 同时设置一个或多个 key-value 对，当且仅当所有给定 key 都不存在
*
* @param maps
* @return 之前已经存在返回false,不存在返回true
*/
public boolean multiSetIfAbsent(Map<String, String> maps) {
return redisTemplate.opsForValue().multiSetIfAbsent(maps);
}
/**
* 增加(自增长), 负数则为自减
*
* @param key
* @return
*/
public Long incrBy(String key, long increment) {
return redisTemplate.opsForValue().increment(key, increment);
}
/**
*
* @param key
* @return
*/
public Double incrByFloat(String key, double increment) {
return redisTemplate.opsForValue().increment(key, increment);
}
/**
* 追加到末尾
*
* @param key
* @param value
* @return
*/
public Integer append(String key, String value) {
return redisTemplate.opsForValue().append(key, value);
}
/** -------------------hash相关操作------------------------- */
/**
* 获取存储在哈希表中指定字段的值
*
* @param key
* @param field
* @return
*/
public Object hGet(String key, String field) {
return redisTemplate.opsForHash().get(key, field);
}
/**
* 获取所有给定字段的值
*
* @param key
* @return
*/
public Map<Object, Object> hGetAll(String key) {
return redisTemplate.opsForHash().entries(key);
}
/**
* 获取所有给定字段的值
*
* @param key
* @param fields
* @return
*/
public List<Object> hMultiGet(String key, Collection<Object> fields) {
return redisTemplate.opsForHash().multiGet(key, fields);
}
public void hPut(String key, String hashKey, String value) {
redisTemplate.opsForHash().put(key, hashKey, value);
}
public void hPutAll(String key, Map<String, String> maps) {
redisTemplate.opsForHash().putAll(key, maps);
}
/**
* 仅当hashKey不存在时才设置
*
* @param key
* @param hashKey
* @param value
* @return
*/
public Boolean hPutIfAbsent(String key, String hashKey, String value) {
return redisTemplate.opsForHash().putIfAbsent(key, hashKey, value);
}
/**
* 删除一个或多个哈希表字段
*
* @param key
* @param fields
* @return
*/
public Long hDelete(String key, Object... fields) {
return redisTemplate.opsForHash().delete(key, fields);
}
/**
* 查看哈希表 key 中，指定的字段是否存在
*
* @param key
* @param field
* @return
*/
public boolean hExists(String key, String field) {
return redisTemplate.opsForHash().hasKey(key, field);
}
/**
* 为哈希表 key 中的指定字段的整数值加上增量 increment
*
* @param key
* @param field
* @param increment
* @return
*/
public Long hIncrBy(String key, Object field, long increment) {
return redisTemplate.opsForHash().increment(key, field, increment);
}
/**
* 为哈希表 key 中的指定字段的整数值加上增量 increment
*
* @param key
* @param field
* @param delta
* @return
*/
public Double hIncrByFloat(String key, Object field, double delta) {
return redisTemplate.opsForHash().increment(key, field, delta);
}
/**
* 获取所有哈希表中的字段
*
* @param key
* @return
*/
public Set<Object> hKeys(String key) {
return redisTemplate.opsForHash().keys(key);
}
/**
* 获取哈希表中字段的数量
*
* @param key
* @return
*/
public Long hSize(String key) {
return redisTemplate.opsForHash().size(key);
}
/**
* 获取哈希表中所有值
*
* @param key
* @return
*/
public List<Object> hValues(String key) {
return redisTemplate.opsForHash().values(key);
}
/**
* 迭代哈希表中的键值对
*
* @param key
* @param options
* @return
*/
public Cursor<Entry<Object, Object>> hScan(String key, ScanOptions options) {
return redisTemplate.opsForHash().scan(key, options);
}
/** ------------------------list相关操作---------------------------- */
/**
* 通过索引获取列表中的元素
*
* @param key
* @param index
* @return
*/
public String lIndex(String key, long index) {
return redisTemplate.opsForList().index(key, index);
}
/**
* 获取列表指定范围内的元素
*
* @param key
* @param start
* 开始位置, 0是开始位置
* @param end
* 结束位置, -1返回所有
* @return
*/
public List<String> lRange(String key, long start, long end) {
return redisTemplate.opsForList().range(key, start, end);
}
/**
* 存储在list头部
*
* @param key
* @param value
* @return
*/
public Long lLeftPush(String key, String value) {
return redisTemplate.opsForList().leftPush(key, value);
}
/**
*
* @param key
* @param value
* @return
*/
public Long lLeftPushAll(String key, String... value) {
return redisTemplate.opsForList().leftPushAll(key, value);
}
/**
*
* @param key
* @param value
* @return
*/
public Long lLeftPushAll(String key, Collection<String> value) {
return redisTemplate.opsForList().leftPushAll(key, value);
}
/**
* 当list存在的时候才加入
*
* @param key
* @param value
* @return
*/
public Long lLeftPushIfPresent(String key, String value) {
return redisTemplate.opsForList().leftPushIfPresent(key, value);
}
/**
* 如果pivot存在,再pivot前面添加
*
* @param key
* @param pivot
* @param value
* @return
*/
public Long lLeftPush(String key, String pivot, String value) {
return redisTemplate.opsForList().leftPush(key, pivot, value);
}
/**
*
* @param key
* @param value
* @return
*/
public Long lRightPush(String key, String value) {
return redisTemplate.opsForList().rightPush(key, value);
}
/**
*
* @param key
* @param value
* @return
*/
public Long lRightPushAll(String key, String... value) {
return redisTemplate.opsForList().rightPushAll(key, value);
}
/**
*
* @param key
* @param value
* @return
*/
public Long lRightPushAll(String key, Collection<String> value) {
return redisTemplate.opsForList().rightPushAll(key, value);
}
/**
* 为已存在的列表添加值
*
* @param key
* @param value
* @return
*/
public Long lRightPushIfPresent(String key, String value) {
return redisTemplate.opsForList().rightPushIfPresent(key, value);
}
/**
* 在pivot元素的右边添加值
*
* @param key
* @param pivot
* @param value
* @return
*/
public Long lRightPush(String key, String pivot, String value) {
return redisTemplate.opsForList().rightPush(key, pivot, value);
}
/**
* 通过索引设置列表元素的值
*
* @param key
* @param index
* 位置
* @param value
*/
public void lSet(String key, long index, String value) {
redisTemplate.opsForList().set(key, index, value);
}
/**
* 移出并获取列表的第一个元素
*
* @param key
* @return 删除的元素
*/
public String lLeftPop(String key) {
return redisTemplate.opsForList().leftPop(key);
}
/**
* 移出并获取列表的第一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止
*
* @param key
* @param timeout
* 等待时间
* @param unit
* 时间单位
* @return
*/
public String lBLeftPop(String key, long timeout, TimeUnit unit) {
return redisTemplate.opsForList().leftPop(key, timeout, unit);
}
/**
* 移除并获取列表最后一个元素
*
* @param key
* @return 删除的元素
*/
public String lRightPop(String key) {
return redisTemplate.opsForList().rightPop(key);
}
/**
* 移出并获取列表的最后一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止
*
* @param key
* @param timeout
* 等待时间
* @param unit
* 时间单位
* @return
*/
public String lBRightPop(String key, long timeout, TimeUnit unit) {
return redisTemplate.opsForList().rightPop(key, timeout, unit);
}
/**
* 移除列表的最后一个元素，并将该元素添加到另一个列表并返回
*
* @param sourceKey
* @param destinationKey
* @return
*/
public String lRightPopAndLeftPush(String sourceKey, String destinationKey) {
return redisTemplate.opsForList().rightPopAndLeftPush(sourceKey,
destinationKey);
}
/**
* 从列表中弹出一个值，将弹出的元素插入到另外一个列表中并返回它； 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止
*
* @param sourceKey
* @param destinationKey
* @param timeout
* @param unit
* @return
*/
public String lBRightPopAndLeftPush(String sourceKey, String destinationKey,
long timeout, TimeUnit unit) {
return redisTemplate.opsForList().rightPopAndLeftPush(sourceKey,
destinationKey, timeout, unit);
}
/**
* 删除集合中值等于value得元素
*
* @param key
* @param index
* index=0, 删除所有值等于value的元素; index>0, 从头部开始删除第一个值等于value的元素;
* index<0, 从尾部开始删除第一个值等于value的元素;
* @param value
* @return
*/
public Long lRemove(String key, long index, String value) {
return redisTemplate.opsForList().remove(key, index, value);
}
/**
* 裁剪list
*
* @param key
* @param start
* @param end
*/
public void lTrim(String key, long start, long end) {
redisTemplate.opsForList().trim(key, start, end);
}
/**
* 获取列表长度
*
* @param key
* @return
*/
public Long lLen(String key) {
return redisTemplate.opsForList().size(key);
}
/** --------------------set相关操作-------------------------- */
/**
* set添加元素
*
* @param key
* @param values
* @return
*/
public Long sAdd(String key, String... values) {
return redisTemplate.opsForSet().add(key, values);
}
/**
* set移除元素
*
* @param key
* @param values
* @return
*/
public Long sRemove(String key, Object... values) {
return redisTemplate.opsForSet().remove(key, values);
}
/**
* 移除并返回集合的一个随机元素
*
* @param key
* @return
*/
public String sPop(String key) {
return redisTemplate.opsForSet().pop(key);
}
/**
* 将元素value从一个集合移到另一个集合
*
* @param key
* @param value
* @param destKey
* @return
*/
public Boolean sMove(String key, String value, String destKey) {
return redisTemplate.opsForSet().move(key, value, destKey);
}
/**
* 获取集合的大小
*
* @param key
* @return
*/
public Long sSize(String key) {
return redisTemplate.opsForSet().size(key);
}
/**
* 判断集合是否包含value
*
* @param key
* @param value
* @return
*/
public Boolean sIsMember(String key, Object value) {
return redisTemplate.opsForSet().isMember(key, value);
}
/**
* 获取两个集合的交集
*
* @param key
* @param otherKey
* @return
*/
public Set<String> sIntersect(String key, String otherKey) {
return redisTemplate.opsForSet().intersect(key, otherKey);
}
/**
* 获取key集合与多个集合的交集
*
* @param key
* @param otherKeys
* @return
*/
public Set<String> sIntersect(String key, Collection<String> otherKeys) {
return redisTemplate.opsForSet().intersect(key, otherKeys);
}
/**
* key集合与otherKey集合的交集存储到destKey集合中
*
* @param key
* @param otherKey
* @param destKey
* @return
*/
public Long sIntersectAndStore(String key, String otherKey, String destKey) {
return redisTemplate.opsForSet().intersectAndStore(key, otherKey,
destKey);
}
/**
* key集合与多个集合的交集存储到destKey集合中
*
* @param key
* @param otherKeys
* @param destKey
* @return
*/
public Long sIntersectAndStore(String key, Collection<String> otherKeys,
String destKey) {
return redisTemplate.opsForSet().intersectAndStore(key, otherKeys,
destKey);
}
/**
* 获取两个集合的并集
*
* @param key
* @param otherKeys
* @return
*/
public Set<String> sUnion(String key, String otherKeys) {
return redisTemplate.opsForSet().union(key, otherKeys);
}
/**
* 获取key集合与多个集合的并集
*
* @param key
* @param otherKeys
* @return
*/
public Set<String> sUnion(String key, Collection<String> otherKeys) {
return redisTemplate.opsForSet().union(key, otherKeys);
}
/**
* key集合与otherKey集合的并集存储到destKey中
*
* @param key
* @param otherKey
* @param destKey
* @return
*/
public Long sUnionAndStore(String key, String otherKey, String destKey) {
return redisTemplate.opsForSet().unionAndStore(key, otherKey, destKey);
}
/**
* key集合与多个集合的并集存储到destKey中
*
* @param key
* @param otherKeys
* @param destKey
* @return
*/
public Long sUnionAndStore(String key, Collection<String> otherKeys,
String destKey) {
return redisTemplate.opsForSet().unionAndStore(key, otherKeys, destKey);
}
/**
* 获取两个集合的差集
*
* @param key
* @param otherKey
* @return
*/
public Set<String> sDifference(String key, String otherKey) {
return redisTemplate.opsForSet().difference(key, otherKey);
}
/**
* 获取key集合与多个集合的差集
*
* @param key
* @param otherKeys
* @return
*/
public Set<String> sDifference(String key, Collection<String> otherKeys) {
return redisTemplate.opsForSet().difference(key, otherKeys);
}
/**
* key集合与otherKey集合的差集存储到destKey中
*
* @param key
* @param otherKey
* @param destKey
* @return
*/
public Long sDifference(String key, String otherKey, String destKey) {
return redisTemplate.opsForSet().differenceAndStore(key, otherKey,
destKey);
}
/**
* key集合与多个集合的差集存储到destKey中
*
* @param key
* @param otherKeys
* @param destKey
* @return
*/
public Long sDifference(String key, Collection<String> otherKeys,
String destKey) {
return redisTemplate.opsForSet().differenceAndStore(key, otherKeys,
destKey);
}
/**
* 获取集合所有元素
*
* @param key
* @return
*/
public Set<String> setMembers(String key) {
return redisTemplate.opsForSet().members(key);
}
/**
* 随机获取集合中的一个元素
*
* @param key
* @return
*/
public String sRandomMember(String key) {
return redisTemplate.opsForSet().randomMember(key);
}
/**
* 随机获取集合中count个元素
*
* @param key
* @param count
* @return
*/
public List<String> sRandomMembers(String key, long count) {
return redisTemplate.opsForSet().randomMembers(key, count);
}
/**
* 随机获取集合中count个元素并且去除重复的
*
* @param key
* @param count
* @return
*/
public Set<String> sDistinctRandomMembers(String key, long count) {
return redisTemplate.opsForSet().distinctRandomMembers(key, count);
}
/**
*
* @param key
* @param options
* @return
*/
public Cursor<String> sScan(String key, ScanOptions options) {
return redisTemplate.opsForSet().scan(key, options);
}
/**------------------zSet相关操作--------------------------------*/
/**
* 添加元素,有序集合是按照元素的score值由小到大排列
*
* @param key
* @param value
* @param score
* @return
*/
public Boolean zAdd(String key, String value, double score) {
return redisTemplate.opsForZSet().add(key, value, score);
}
/**
*
* @param key
* @param values
* @return
*/
public Long zAdd(String key, Set<TypedTuple<String>> values) {
return redisTemplate.opsForZSet().add(key, values);
}
/**
*
* @param key
* @param values
* @return
*/
public Long zRemove(String key, Object... values) {
return redisTemplate.opsForZSet().remove(key, values);
}
/**
* 增加元素的score值，并返回增加后的值
*
* @param key
* @param value
* @param delta
* @return
*/
public Double zIncrementScore(String key, String value, double delta) {
return redisTemplate.opsForZSet().incrementScore(key, value, delta);
}
/**
* 返回元素在集合的排名,有序集合是按照元素的score值由小到大排列
*
* @param key
* @param value
* @return 0表示第一位
*/
public Long zRank(String key, Object value) {
return redisTemplate.opsForZSet().rank(key, value);
}
/**
* 返回元素在集合的排名,按元素的score值由大到小排列
*
* @param key
* @param value
* @return
*/
public Long zReverseRank(String key, Object value) {
return redisTemplate.opsForZSet().reverseRank(key, value);
}
/**
* 获取集合的元素, 从小到大排序
*
* @param key
* @param start
* 开始位置
* @param end
* 结束位置, -1查询所有
* @return
*/
public Set<String> zRange(String key, long start, long end) {
return redisTemplate.opsForZSet().range(key, start, end);
}
/**
* 获取集合元素, 并且把score值也获取
*
* @param key
* @param start
* @param end
* @return
*/
public Set<TypedTuple<String>> zRangeWithScores(String key, long start,
long end) {
return redisTemplate.opsForZSet().rangeWithScores(key, start, end);
}
/**
* 根据Score值查询集合元素
*
* @param key
* @param min
* 最小值
* @param max
* 最大值
* @return
*/
public Set<String> zRangeByScore(String key, double min, double max) {
return redisTemplate.opsForZSet().rangeByScore(key, min, max);
}
/**
* 根据Score值查询集合元素, 从小到大排序
*
* @param key
* @param min
* 最小值
* @param max
* 最大值
* @return
*/
public Set<TypedTuple<String>> zRangeByScoreWithScores(String key,
double min, double max) {
return redisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max);
}
/**
*
* @param key
* @param min
* @param max
* @param start
* @param end
* @return
*/
public Set<TypedTuple<String>> zRangeByScoreWithScores(String key,
double min, double max, long start, long end) {
return redisTemplate.opsForZSet().rangeByScoreWithScores(key, min, max,
start, end);
}
/**
* 获取集合的元素, 从大到小排序
*
* @param key
* @param start
* @param end
* @return
*/
public Set<String> zReverseRange(String key, long start, long end) {
return redisTemplate.opsForZSet().reverseRange(key, start, end);
}
/**
* 获取集合的元素, 从大到小排序, 并返回score值
*
* @param key
* @param start
* @param end
* @return
*/
public Set<TypedTuple<String>> zReverseRangeWithScores(String key,
long start, long end) {
return redisTemplate.opsForZSet().reverseRangeWithScores(key, start,
end);
}
/**
* 根据Score值查询集合元素, 从大到小排序
*
* @param key
* @param min
* @param max
* @return
*/
public Set<String> zReverseRangeByScore(String key, double min,
double max) {
return redisTemplate.opsForZSet().reverseRangeByScore(key, min, max);
}
/**
* 根据Score值查询集合元素, 从大到小排序
*
* @param key
* @param min
* @param max
* @return
*/
public Set<TypedTuple<String>> zReverseRangeByScoreWithScores(
String key, double min, double max) {
return redisTemplate.opsForZSet().reverseRangeByScoreWithScores(key,
min, max);
}
/**
*
* @param key
* @param min
* @param max
* @param start
* @param end
* @return
*/
public Set<String> zReverseRangeByScore(String key, double min,
double max, long start, long end) {
return redisTemplate.opsForZSet().reverseRangeByScore(key, min, max,
start, end);
}
/**
* 根据score值获取集合元素数量
*
* @param key
* @param min
* @param max
* @return
*/
public Long zCount(String key, double min, double max) {
return redisTemplate.opsForZSet().count(key, min, max);
}
/**
* 获取集合大小
*
* @param key
* @return
*/
public Long zSize(String key) {
return redisTemplate.opsForZSet().size(key);
}
/**
* 获取集合大小
*
* @param key
* @return
*/
public Long zZCard(String key) {
return redisTemplate.opsForZSet().zCard(key);
}
/**
* 获取集合中value元素的score值
*
* @param key
* @param value
* @return
*/
public Double zScore(String key, Object value) {
return redisTemplate.opsForZSet().score(key, value);
}
/**
* 移除指定索引位置的成员
*
* @param key
* @param start
* @param end
* @return
*/
public Long zRemoveRange(String key, long start, long end) {
return redisTemplate.opsForZSet().removeRange(key, start, end);
}
/**
* 根据指定的score值的范围来移除成员
*
* @param key
* @param min
* @param max
* @return
*/
public Long zRemoveRangeByScore(String key, double min, double max) {
return redisTemplate.opsForZSet().removeRangeByScore(key, min, max);
}
/**
* 获取key和otherKey的并集并存储在destKey中
*
* @param key
* @param otherKey
* @param destKey
* @return
*/
public Long zUnionAndStore(String key, String otherKey, String destKey) {
return redisTemplate.opsForZSet().unionAndStore(key, otherKey, destKey);
}
/**
*
* @param key
* @param otherKeys
* @param destKey
* @return
*/
public Long zUnionAndStore(String key, Collection<String> otherKeys,
String destKey) {
return redisTemplate.opsForZSet()
.unionAndStore(key, otherKeys, destKey);
}
/**
* 交集
*
* @param key
* @param otherKey
* @param destKey
* @return
*/
public Long zIntersectAndStore(String key, String otherKey,
String destKey) {
return redisTemplate.opsForZSet().intersectAndStore(key, otherKey,
destKey);
}
/**
* 交集
*
* @param key
* @param otherKeys
* @param destKey
* @return
*/
public Long zIntersectAndStore(String key, Collection<String> otherKeys,
String destKey) {
return redisTemplate.opsForZSet().intersectAndStore(key, otherKeys,
destKey);
}
/**
*
* @param key
* @param options
* @return
*/
public Cursor<TypedTuple<String>> zScan(String key, ScanOptions options) {
return redisTemplate.opsForZSet().scan(key, options);
}
}
6 Redis配置文件详解
6.1 容量单位
不区分大小写，G和GB有区别
6.2 使用多个配置
使用 include 组合
6.3 网络配置
6.4 通用GENERAL
6.5日志输出级别
6.6 日志输出文件
6.7 持久化规则(快照)
持久化方式：
- RDB
- AOF
6.8 RDB文件相关
6.9 安全
6.10 限制
6.11 APPEND ONLY 模式 aof配置
7 redis持久化
7.1 RDB
RDB：Redis Databases
7.1.1 什么是RDB
在指定时间间隔后，将内存中的数据集快照写入数据库 ；在恢复时候，直接读取快照文件，进行数据的恢复 ；
默认情况下， Redis 将数据库快照保存在名字为 dump.rdb的二进制文件中。文件名可以在配置文件中进行自定义。
7.1.2 工作原理
在进行 RDB 的时候，redis 的主线程是不会做 io 操作的，主线程会 fork 一个子线程来完成该操作；
-
Redis 调用forks。同时拥有父进程和子进程。
-
子进程将数据集写入到一个临时 RDB 文件中。
-
当子进程完成对新 RDB 文件的写入时，Redis 用新 RDB 文件替换原来的 RDB 文件，并删除旧的 RDB 文件。
这种工作方式使得 Redis 可以从写时复制（copy-on-write）机制中获益(因为是使用子进程进行写操作，而父进程依然可以接收来自客户端的请求。)
7.1.3 触发机制
- save的规则满足的情况下，会自动触发rdb原则
- 执行flushall命令，也会触发我们的rdb原则
- 退出redis，也会自动产生rdb文件
使用 save 命令，会立刻对当前内存中的数据进行持久化 ,但是会阻塞，也就是不接受其他操作了；
由于 save 命令是同步命令，会占用Redis的主进程。若Redis数据非常多时，save命令执行速度会非常慢，阻塞所有客户端的请求。
flushall 命令也会触发持久化 ；
bgsave 是异步进行，进行持久化的时候，redis 还可以将继续响应客户端请求 ；
bgsave和save对比
| 命令 | save | bgsave |
|---|---|---|
| IO类型 | 同步 | 异步 |
| 阻塞？ | 是 | 是（阻塞发生在fock()，通常非常快） |
| 复杂度 | O(n) | O(n) |
| 优点 | 不会消耗额外的内存 | 不阻塞客户端命令 |
| 缺点 | 阻塞客户端命令 | 需要fock子进程，消耗内存 |
RDB优点：
- 适合大规模的数据恢复
- 对数据的完整性要求不高
RDB缺点：
- 需要一定的时间间隔进行操作，如果redis意外宕机了，这个最后一次修改的数据就没有了。
- fork进程的时候，会占用一定的内容空间。
7.1.4 触发持久化规则
可以通过配置文件对 Redis 进行设置， 让它在“ N 秒内数据集至少有 M 个改动”这一条件被满足时， 自动进行数据集保存操作。
7.2 AOF
Append Only File
将我们所有的命令都记录下来，history，恢复的时候就把这个文件全部再执行一遍
7.2.1 什么是AOF
以日志的形式来记录每个写的操作，将Redis执行过的所有指令记录下来（读操作不记录），只许追加文件但不可以改写文件，redis启动之初会读取该文件重新构建数据，换言之，redis重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作。
快照功能（RDB）并不是非常耐久（durable）： 如果 Redis 因为某些原因而造成故障停机， 那么服务器将丢失最近写入、以及未保存到快照中的那些数据。 从 1.1 版本开始， Redis 增加了一种完全耐久的持久化方式： AOF 持久化。
如果要使用AOF，需要修改配置文件：
appendonly no yes则表示启用AOF
默认是不开启的，我们需要手动配置，然后重启redis，就可以生效了！
如果这个aof文件有错位，这时候redis是启动不起来的，我需要修改这个aof文件
redis给我们提供了一个工具redis-check-aof --fix
7.2.2 aof优缺点
appendonly yes # 默认是不开启aof模式的，默认是使用rdb方式持久化的，在大部分的情况下，rdb完全够用
appendfilename "appendonly.aof"
# appendfsync always # 每次修改都会sync 消耗性能
appendfsync everysec # 每秒执行一次 sync 可能会丢失这一秒的数据
# appendfsync no # 不执行 sync ,这时候操作系统自己同步数据，速度最快
优点
- 每一次修改都会同步，文件的完整性会更加好
- 没秒同步一次，可能会丢失一秒的数据
- 从不同步，效率最高
缺点
- 相对于数据文件来说，aof远远大于rdb，修复速度比rdb慢！
- Aof运行效率也要比rdb慢，所以我们redis默认的配置就是rdb持久化
7.3 RDB和AOF的对比和选择
| RDB | AOF | |
|---|---|---|
| 启动优先级 | 低 | 高 |
| 体积 | 小 | 大 |
| 恢复速度 | 快 | 慢 |
| 数据安全性 | 丢数据 | 根据策略决定 |
如何选择使用哪种持久化方式？
一般来说， 如果想达到足以媲美 PostgreSQL 的数据安全性， 你应该同时使用两种持久化功能。
如果你非常关心你的数据， 但仍然可以承受数分钟以内的数据丢失， 那么你可以只使用 RDB 持久化。
有很多用户都只使用 AOF 持久化， 但并不推荐这种方式： 因为定时生成 RDB 快照（snapshot）非常便于进行数据库备份， 并且 RDB 恢复数据集的速度也要比 AOF 恢复的速度要快。
8 Redis发布和订阅
Redis 发布订阅(pub/sub)是一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息。
下图展示了频道 channel1 ， 以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系：
当有新消息通过 PUBLISH 命令发送给频道 channel1 时， 这个消息就会被发送给订阅它的三个客户端：
8.1 相关命令
| 命令 | 描述 |
|---|---|
| PSUBSCRIBE pattern [pattern…] | 订阅一个或多个符合给定模式的频道。 |
| PUNSUBSCRIBE pattern [pattern…] | 退订一个或多个符合给定模式的频道。 |
| PUBSUB subcommand [argument[argument]] | 查看订阅与发布系统状态。 |
| PUBLISH channel message | 向指定频道发布消息 |
| SUBSCRIBE channel [channel…] | 订阅给定的一个或多个频道。 |
| SUBSCRIBE channel [channel…] | 退订一个或多个频道 |
8.2 实例
------------订阅端----------------------
127.0.0.1:6379> SUBSCRIBE sakura # 订阅sakura频道
Reading messages... (press Ctrl-C to quit) # 等待接收消息
1) "subscribe" # 订阅成功的消息
2) "sakura"
3) (integer) 1
1) "message" # 接收到来自sakura频道的消息 "hello world"
2) "sakura"
3) "hello world"
1) "message" # 接收到来自sakura频道的消息 "hello i am sakura"
2) "sakura"
3) "hello i am sakura"
--------------消息发布端-------------------
127.0.0.1:6379> PUBLISH sakura "hello world" # 发布消息到sakura频道
(integer) 1
127.0.0.1:6379> PUBLISH sakura "hello i am sakura" # 发布消息
(integer) 1
-----------------查看活跃的频道------------
127.0.0.1:6379> PUBSUB channels
1) "sakura"
8.3 原理
每个 Redis 服务器进程都维持着一个表示服务器状态的 redis.h/redisServer 结构， 结构的 pubsub_channels 属性是一个字典， 这个字典就用于保存订阅频道的信息，其中，字典的键为正在被订阅的频道， 而字典的值则是一个链表， 链表中保存了所有订阅这个频道的客户端。
客户端订阅，就被链接到对应频道的链表的尾部，退订则就是将客户端节点从链表中移除。
8.4 缺点及应用
缺点:
如果一个客户端订阅了频道，但自己读取消息的速度却不够快的话，那么不断积压的消息会使redis输出缓冲区的体积变得越来越大，这可能使得redis本身的速度变慢，甚至直接崩溃。
这和数据传输可靠性有关，如果在订阅方断线，那么他将会丢失所有在短线期间发布者发布的消息。
应用:
消息订阅：公众号订阅，微博关注等等（起始更多是使用消息队列来进行实现）
多人在线聊天室。
稍微复杂的场景，我们就会使用消息中间件MQ处理。
9 主从复制
9.1 概念
主从复制，是指将一台Redis服务器的数据，复制到其他的Redis服务器。前者称为主节点（Master/Leader）,后者称为从节点（Slave/Follower）， 数据的复制是单向的！只能由主节点复制到从节点（主节点以写为主、从节点以读为主）。默认情况下，每台Redis服务器都是主节点，一个主节点可以有0个或者多个从节点，但每个从节点只能由一个主节点。
9.2 作用
数据冗余：主从复制实现了数据的热备份，是持久化之外的一种数据冗余的方式。
故障恢复：当主节点故障时，从节点可以暂时替代主节点提供服务，是一种服务冗余的方式
负载均衡：在主从复制的基础上，配合读写分离，由主节点进行写操作，从节点进行读操作，分担服务器的负载；尤其是在多读少写的场景下，通过多个从节点分担负载，提高并发量。
高可用基石：主从复制还是哨兵和集群能够实施的基础。
9.3 为什么使用集群
- 单台服务器难以负载大量的请求
- 单台服务器故障率高，系统崩坏概率大
- 单台服务器内存容量有限。
9.4 环境配置
我们在讲解配置文件的时候，注意到有一个replication模块 (见Redis.conf中第8条)
查看当前库的信息：info replication
127.0.0.1:6379> info replication
# Replication
role:master # 角色
connected_slaves:0 # 从机数量
master_replid:3b54deef5b7b7b7f7dd8acefa23be48879b4fcff
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0
既然需要启动多个服务，就需要多个配置文件。每个配置文件对应修改以下信息：
- 端口号
- pid文件名
- 日志文件名
- rdb文件名
启动单机多服务集群：
9.5 一主二从配置
==默认情况下，每台Redis服务器都是主节点；==我们一般情况下只用配置从机就好了！
认老大！一主（79）二从（80，81）
使用SLAVEOF host port就可以为从机配置主机了。
然后主机上也能看到从机的状态：
我们这里是使用命令搭建，是暂时的，==真实开发中应该在从机的配置文件中进行配置，==这样的话是永久的。
9.6 使用规则
从机只能读，不能写，主机可读可写但是多用于写。
127.0.0.1:6381> set name sakura # 从机6381写入失败
(error) READONLY You can't write against a read only replica.
127.0.0.1:6380> set name sakura # 从机6380写入失败
(error) READONLY You can't write against a read only replica.
127.0.0.1:6379> set name sakura
OK
127.0.0.1:6379> get name
"sakura"
当主机断电宕机后，默认情况下从机的角色不会发生变化 ，集群中只是失去了写操作，当主机恢复以后，又会连接上从机恢复原状。
当从机断电宕机后，若不是使用配置文件配置的从机，再次启动后作为主机是无法获取之前主机的数据的，若此时重新配置称为从机，又可以获取到主机的所有数据。这里就要提到一个同步原理。
第二条中提到，默认情况下，主机故障后，不会出现新的主机，有两种方式可以产生新的主机：
从机手动执行命令slaveof no one,这样执行以后从机会独立出来成为一个主机
使用哨兵模式（自动选举）
如果没有老大了，这个时候能不能选择出来一个老大呢？手动！
如果主机断开了连接，我们可以使用SLAVEOF no one让自己变成主机！其他的节点就可以手动连接到最新的主节点（手动）！如果这个时候老大修复了，那么久重新连接！
10 哨兵模式
更多信息参考博客：https://www.jianshu.com/p/06ab9daf921d
**主从切换技术的方法是：当主服务器宕机后，需要手动把一台从服务器切换为主服务器，这就需要人工干预，费事费力，还会造成一段时间内服务不可用。这不是一种推荐的方式，更多时候，我们优先考虑哨兵模式。
10.1 哨兵的作用：
- 通过发送命令，让Redis服务器返回监控其运行状态，包括主服务器和从服务器。
- 当哨兵监测到master宕机，会自动将slave切换成master，然后通过发布订阅模式通知其他的从服务器，修改配置文件，让它们切换主机。
10.2 哨兵的核心配置
sentinel monitor mymaster 127.0.0.1 6379 1
- 数字1表示 ：当一个哨兵主观认为主机断开，就可以客观认为主机故障，然后开始选举新的主机。
10.3测试
redis-sentinel xxx/sentinel.conf
成功启动哨兵模式
此时哨兵监视着我们的主机6379，当我们断开主机后：
10.4 哨兵模式优缺点
优点：
哨兵集群，基于主从复制模式，所有主从复制的优点，它都有
主从可以切换，故障可以转移，系统的可用性更好
哨兵模式是主从模式的升级，手动到自动，更加健壮
缺点：
Redis不好在线扩容，集群容量一旦达到上限，在线扩容就十分麻烦
实现哨兵模式的配置其实是很麻烦的，里面有很多配置项
10.5 哨兵模式的全部配置
完整的哨兵模式配置文件 sentinel.conf
# Example sentinel.conf
# 哨兵sentinel实例运行的端口 默认26379
port 26379
# 哨兵sentinel的工作目录
dir /tmp
# 哨兵sentinel监控的redis主节点的 ip port
# master-name 可以自己命名的主节点名字 只能由字母A-z、数字0-9 、这三个字符".-_"组成。
# quorum 当这些quorum个数sentinel哨兵认为master主节点失联 那么这时 客观上认为主节点失联了
# sentinel monitor <master-name> <ip> <redis-port> <quorum>
sentinel monitor mymaster 127.0.0.1 6379 1
# 当在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>
sentinel failover-timeout mymaster 180000
# SCRIPTS EXECUTION
#配置当某一事件发生时所需要执行的脚本，可以通过脚本来通知管理员，例如当系统运行不正常时发邮件通知相关人员。
#对于脚本的运行结果有以下规则：
#若脚本执行后返回1，那么该脚本稍后将会被再次执行，重复次数目前默认为10
#若脚本执行后返回2，或者比2更高的一个返回值，脚本将不会重复执行。
#如果脚本在执行过程中由于收到系统中断信号被终止了，则同返回值为1时的行为相同。
#一个脚本的最大执行时间为60s，如果超过这个时间，脚本将会被一个SIGKILL信号终止，之后重新执行。
#通知型脚本:当sentinel有任何警告级别的事件发生时（比如说redis实例的主观失效和客观失效等等），将会去调用这个脚本，
#这时这个脚本应该通过邮件，SMS等方式去通知系统管理员关于系统不正常运行的信息。调用该脚本时，将传给脚本两个参数，
#一个是事件的类型，
#一个是事件的描述。
#如果sentinel.conf配置文件中配置了这个脚本路径，那么必须保证这个脚本存在于这个路径，并且是可执行的，否则sentinel无法正常启动成功。
#通知脚本
# 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
11 缓存穿透和雪崩
11.1 缓存穿透
概念
在默认情况下，用户请求数据时，会先在缓存(Redis)中查找，若没找到即缓存未命中，再在数据库中进行查找，数量少可能问题不大，可是一旦大量的请求数据（例如秒杀场景）缓存都没有命中的话，就会全部转移到数据库上，造成数据库极大的压力，就有可能导致数据库崩溃。网络安全中也有人恶意使用这种手段进行攻击被称为洪水攻击。
缓存穿透的解决方案:
1.布隆过滤器
对所有可能查询的参数以Hash的形式存储，以便快速确定是否存在这个值，在控制层先进行拦截校验，校验不通过直接打回，减轻了存储系统的压力。
2.缓存空对象
一次请求若在缓存和数据库中都没找到，就在缓存中方一个空对象用于处理后续这个请求。
这样做有一个缺陷：存储空对象也需要空间，大量的空对象会耗费一定的空间，存储效率并不高。解决这个缺陷的方式就是设置较短过期时间
即使对空值设置了过期时间，还是会存在缓存层和存储层的数据会有一段时间窗口的不一致，这对于需要保持一致性的业务会有影响。
11.2 缓存击穿
概念
相较于缓存穿透，缓存击穿的目的性更强，一个存在的key，在缓存过期的一刻，同时有大量的请求，这些请求都会击穿到DB，造成瞬时DB请求量大、压力骤增。这就是缓存被击穿，只是针对其中某个key的缓存不可用而导致击穿，但是其他的key依然可以使用缓存响应。
比如热搜排行上，一个热点新闻被同时大量访问就可能导致缓存击穿。
解决方案
1.设置热点数据永不过期
这样就不会出现热点数据过期的情况，但是当Redis内存空间满的时候也会清理部分数据，而且此种方案会占用空间，一旦热点数据多了起来，就会占用部分空间。
2.加互斥锁(分布式锁)
在访问key之前，采用SETNX（set if not exists）来设置另一个短期key来锁住当前key的访问，访问结束再删除该短期key。保证同时刻只有一个线程访问。这样对锁的要求就十分高。
11.3 缓存雪崩
概念
大量的key设置了相同的过期时间，导致在缓存在同一时刻全部失效，造成瞬时DB请求量大、压力骤增，引起雪崩。
解决方案
- redis高可用
这个思想的含义是，既然redis有可能挂掉，那我多增设几台redis，这样一台挂掉之后其他的还可以继续工作，其实就是搭建的集群
- 限流降级
这个解决方案的思想是，在缓存失效后，通过加锁或者队列来控制读数据库写缓存的线程数量。比如对某个key只允许一个线程查询数据和写缓存，其他线程等待。
- 数据预热
数据加热的含义就是在正式部署之前，我先把可能的数据先预先访问一遍，这样部分可能大量访问的数据就会加载到缓存中。在即将发生大并发访问前手动触发加载缓存不同的key，设置不同的过期时间，让缓存失效的时间点尽量均匀。
12 学习链接
参考链接：https://blog.csdn.net/DDDDeng_/article/details/108118544?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522163032815816780261933185%2522%252C%2522scm%2522%253A%252220140713.130102334…%2522%257D&request_id=163032815816780261933185&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2allsobaiduend~default-1-108118544.first_rank_v2_pc_rank_v29&utm_term=redis%E7%8B%82%E7%A5%9E%E8%AF%B4%E7%AC%94%E8%AE%B0&spm=1018.2226.3001.4187
狂神视频链接：https://www.bilibili.com/video/BV1S54y1R7SB?p=36
