Redis Master& Slave + Sentinel+ Lettuce 二
主从复制(Master& Slave)作用同步过程/ 复制原理具体过程如下：
Redis配置基本参数RDB持久化相关参数REPLICATION(复制/主从同步)相关参数SECURITY(安全)相关参数AOF(Append Only File)持久化相关参数LUA脚本SLOW LOG(慢查询日志)脚本多服务单机简单部署
哨兵模式(Reids Sentinel)作用原理& 过程Redis Sentinel配置多服务单机简单部署
Spring boot Lettuce使用Jar包配置实例
主从复制(Master& Slave)
作用
数据备份: 除 RDB& AOF(持久化)以外的数据备份方式故障恢复: 当主节点(Master)出现问题时, 可以将从节点(Slave)用于主节点, 及时的恢复服务, 实现服务的高可用读写分离: 主节点写, 从节点读, 实现读写分离, 提高服务器的负载能力
除此之外, 主从复制也是 Redis集群的基础
同步过程/ 复制原理
全量同步: 发生在从节点初始化阶段, 将主节点的所有数据做成快照(RDB)文件, 发送给从节点过程; *当数据量较大时, 会对主从节点和网络带宽造成很大的开销
增量同步: Redis的增量同步是指从节点完成初始化后开始正常工作时, 主节点将新产生的写操作命令同步到从节点, 然后在从节点执行的过程
具体过程如下：
从节点发起 SYNC命令到主节点主节点收到 SYNC命令后, 执行 BGSAVE命令, 在异步进程内生成快照(RDB)文件, 并使用缓冲区记录同时阶段产生的所有写操作命令主节点生成完快照(RDB)文件后, 将该文件发送到从节点, 在此发送期间产生的所有写操作命令继续记录从节点收到快照(RDB)文件后, 丢弃所有旧数据, 并载入收到的快照主节点发送完快照文件后, 开始向从节点发送缓冲区内记录的(全量同步期间额外记录的)写操作命令从节点完成对快照的载入后, 开始接收来自主节点的操作命令并执行
完成全量同步后, 如无特殊情况一直是增量同步, 就比如从节点断线服务中断一段时间后重启, Redis会尽可能首先尝试增量同步, 否者, 全量同步. 还有主节点, 因同步时需要开辟额外的缓冲区来记录命令等操作, 所以实际场景中得考虑好主节点的内存剩余空间
Redis单台服务默认是主节点, 通过命令方式或配置方式, 可以指向其它节点(被指向的节点可以是主也可以是从), 同时当前节点会变成从节点; 主节点可以有多个从节点, 但从节点只能有一个主节点; 数据的复制是单向的, 只能主节点到从节点
Redis配置
基本参数
# 外部网络是否可以连接服务 yes启用(默认)/ no
protected-mode yes
# Redis服务端口
port 6379
# 是否守护进程运行 yes/ no(默认)
daemonize no
# 是否通过 supervised管理守护进程 no不使用(默认)/ upstart/ systemd/ auto
supervised no
# Redis进程编号存储文件& path
pidfile "/var/run/redis_6379.pid"
RDB持久化相关参数
https://blog.csdn.net/qcl108/article/details/106879002#RDB_AOF_36
# RDB持久化文件名
dbfilename "dump_6379.rdb"
# 数据库存入目录
dir "/Users/quanchunlin/Desktop/redis-6.0.5/conf"
REPLICATION(复制/主从同步)相关参数
# 配置形式指向节点与命令行的 slaveof相同的效果
replicaof <masterip> <masterport>
# 设置主节点的密码
masterauth <master-password>
# 当前节点为从节点时, 将会无法执行写操作命令 yes(默认)/ no
replica-read-only yes
# 主从同步策略: disk与 socket(diskless/无磁盘方式 此方式还处于实验阶段). yes/ no不启用(默认)
repl-diskless-sync no
# 从节点指定往主节点发送 Ping的时间间隔. 10秒(默认)
# repl-ping-replica-period 10
# 同步的超时时间
# repl-timeout 60
# 是否启用 TCP_NODELAY, no不开启(默认)/yes. 如果开启则会使用少量的 TCP包进行数据传输到从节点, 速度会比较慢; 如不开启传输速度会比较快, 但会占用较多的带宽
repl-disable-tcp-nodelay no
# 设置缓冲区大小, 在从节点失连时主节点存放要同步到从节点的数据, 设置的越大, 从节点可以失连的时间就越长
# repl-backlog-size 1mb
# 如果一段时间内没有从节点连接主节点, 则会释放缓冲区. 设置0则表示永不释放缓冲区
# repl-backlog-ttl 3600
# 当主节点无法工作时哨兵(Sentinel)通过这个值来决定将哪个从(Slave)优先提升为主节点, 值越小表示越优先提升, 但设置为0表示, 此节点将永远不能被提升为主节点. 100(默认)
replica-priority 100
# 当 Master少于3个, 延时小于等于10秒的已连接 Slave, 就可以停止接收写操作
# 1. 至少需要3个 Slave的状态为 oneline
# 2. 延时是以秒为单位, 且必须小于等于指定值, 是从最后一个 Slave接收到的 ping(通常每秒发送)开始计数
# min-replicas-to-write 3
# min-replicas-max-lag 10
SECURITY(安全)相关参数
# 当前服务节点密码
# requirepass foobared
# 命令重命名: 用于危险命令改变名字. 注: 该命令记录到 AOF文件后被传送到从服务器可能产生问题
# rename-command CONFIG ""
AOF(Append Only File)持久化相关参数
https://blog.csdn.net/qcl108/article/details/106879002#RDB_AOF_36
# yes开启/ no关闭(默认)
appendonly no
# AOF文件名 (default: "appendonly.aof")
appendfilename "appendonly_6379.aof"
LUA脚本
# Lua脚本的最大执行时间, 毫秒为单位
lua-time-limit 5000
SLOW LOG(慢查询日志)脚本
# 记录超过多少微秒的查询命令. 查询的执行时间不包括客户端的 I/O执行和网络通信时间, 只是查询命令执行时间
# 1000000等于1秒, 设置为0则记录所有命令
slowlog-log-slower-than 10000
# 记录大小，可通过SLOWLOG RESET命令重置
slowlog-max-len 128
多服务单机简单部署
创建三个 redis.conf环境, 1个 Master, 2个 Slave. 放到同一个目录
# 主节点 redis_6379.conf
port 6379
dbfilename dump_6379.rdb
# 从节点 redis_6380.conf
port 6380
dbfilename dump_6380.rdb
replicaof 127.0.0.1 6379 # 可以启动后手动在 redis-cli, 通过 slaveof命令指向主节点
# 从节点 redis_6381.conf
port 6381
dbfilename dump_6381.rdb
replicaof 127.0.0.1 6379 # 可以启动后手动在 redis-cli, 通过 slaveof命令指向主节点
依次启动: …/src/redis-server redis-6379.conf, …/src/redis-server redis-6380.conf, …/src/redis-server redis-6381.conf启动后可以通过 info replication命令查看主从信息
哨兵模式(Reids Sentinel)
作用
在多从单主架构中, 一旦主节点由于故障不能提供服务时, 通过哨兵(Sentinel)模式, 将其中一个从节点自动晋升为主节点, 同时通知其它从节点重新指向新的主节点, 来维持高可用 Redis服务
原理& 过程
主观下线
每隔一秒, 每个 Sentinel会向主节点或从节点做心跳检测, 当一个主节点心跳反应超过 sentinel down-after-milliseconds设的毫秒数, Sentinel将会对该节点做失败判定并标注
客观下线
当主观下线的节点为主节点时, 该 Sentinel会向其它 Sentinel询问对主节点的判断, 当至少 sentinel monitor的数 Sentinel同意判定时, 该 Sentinel会做出客观下线的决定
领导者 Sentinel选举
Raft算法实现: 假设 s1(sentinel-1)最先完成客观下线, 它会向其余 Sentinel发送命令, 请求成为领导者, 收到请求的 Sentinel如果没有同意过其它 Sentinel的请求, 那么就会同意 s1的请求, 成为领导者
故障转移
领导者 Sentinel在多个从节点中选出一个节点作为新的主节点, 选与前主节点数据相似度最高的从节点, 然后将其它从节点重新指向新的主节点, Sentinel集群也会将原主节点改为从节点并继续监控, 当其恢复后命令它去复制新的主节点
Redis Sentinel配置
# 外部网络是否可以连接服务 yes启用/ no(默认)
protected-mode no
# Redis Sentinel服务端口
port 26379
# 是否守护进程运行 yes/ no(默认)
daemonize no
# Redis Sentinel进程编号存储文件& path
pidfile "/var/run/redis-sentinel-26379.pid"
# 日志文件
logfile "/var/log/sentinel-26379.log"
# 工作目录
dir "/Users/quanchunlin/Desktop/redis-6.0.5/conf"
# 设置<master-name>标识集群名称可以自定义, 设置<password>连接主从节点时的密码; 注: Sentinel无法为主从分别设置密码, 所以需将密码都设置一样的
# sentinel auth-pass <master-name> <password>
# 监听地址为 <ip> <redis-port>的主节点, <quorum>数是有多少个 Sentinel同意当前主节点失效时, 才会被当前 Sentinel集群认为是客观下线
sentinel monitor <master-name> <ip> <redis-port> <quorum>
# 当主备切换时, 某台从节点晋升为新的主节点时, 同时其它从节点被重新指向新的主节点做主从同步, <numreplicas>值越小同步完成时间越长, 但如果值设的过大, 可能会导致主节点阻塞
# sentinel parallel-syncs <master-name> <numreplicas>
# Specifies the failover timeout in milliseconds. It is used in many ways:
#
# - The time needed to re-start a failover after a previous failover was
# already tried against the same master by a given Sentinel, is two
# times the failover timeout.
#
# - The time needed for a replica replicating to a wrong master according
# to a Sentinel current configuration, to be forced to replicate
# with the right master, is exactly the failover timeout (counting since
# the moment a Sentinel detected the misconfiguration).
#
# - The time needed to cancel a failover that is already in progress but
# did not produced any configuration change (SLAVEOF NO ONE yet not
# acknowledged by the promoted replica).
#
# - The maximum time a failover in progress waits for all the replicas to be
# reconfigured as replicas of the new master. However even after this time
# the replicas will be reconfigured by the Sentinels anyway, but not with
# the exact parallel-syncs progression as specified.
# sentinel failover-timeout <master-name> <milliseconds>
# 心跳反应需要超过多少失效时间, Sentinel才将一个主节点主观的地判定是不可用的. 30000毫秒(默认)
sentinel down-after-milliseconds <master-name> <milliseconds>
# 通知脚本: 当有警告级别的事件发生时(比如说 Redis实例的主观失效和客观失效等)将会调用此脚本, 一般用于发送邮件或 SMS等方式通知系统管理员. 调用该脚本时, 将传给脚本两个参数, 一个是事件的类型, 一个是事件的描述
# sentinel notification-script <master-name> <script-path>
# 当主备切换时, 将会自动调用此脚本. 调用该脚本时, 将传给脚本以下参数 <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port>
# sentinel client-reconfig-script <master-name> <script-path>
多服务单机简单部署
创建三个 sentinel.conf环境
# sentinel-26379.conf
port 26379
sentinel monitor mymaster 127.0.0.1 6379 2
sentinel down-after-milliseconds mymaster 10000
# sentinel-26380.conf
port 26380
sentinel monitor mymaster 127.0.0.1 6379 2
sentinel down-after-milliseconds mymaster 20000
# sentinel-26381.conf
port 26381
sentinel monitor mymaster 127.0.0.1 6379 2
sentinel down-after-milliseconds mymaster 30000
依次启动: …/src/redis-sentinel sentinel-26379.conf, …/src/redis-sentinel sentinel-26380.conf, …/src/redis-sentinel sentinel-26381.conf
Spring boot Lettuce使用
Jar包
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
配置
# 使用的数据库索引编号
spring.redis.database=0
# 连接池最大连接数(-1表示不限制), 默认为8
spring.redis.lettuce.pool.max-active=1000
# 连接池最大阻塞等待时间(-1表示不限制, 单位毫秒ms), 默认为-1
spring.redis.lettuce.pool.max-wait=-1
# 连接池中的最大空闲连接, 默认为8
spring.redis.lettuce.pool.max-idle=200
# 连接池中的最小空闲连接, 默认为0
spring.redis.lettuce.pool.min-idle=100
# 关闭连接前等待任务处理完成的最长时间, 默认为100ms
spring.redis.lettuce.shutdown-timeout=100ms
# 哨兵设定的主节点名称
spring.redis.sentinel.master=mymaster
# 哨兵节点, 多节点按(,)号分割
spring.redis.sentinel.nodes=127.0.0.1:26379,127.0.0.1:26380,127.0.0.1:26381
实例
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate redisTemplate(RedisConnectionFactory factory) {
final RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(factory);
/** 将默认 Jdk序列化替换成 StringRedisSerializer*/
final StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
template.setKeySerializer(stringRedisSerializer);
template.setHashKeySerializer(stringRedisSerializer);
/** 将默认 Jdk序列化替换成 Jackson2JsonRedisSerialize*/
final Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
final ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
template.setValueSerializer(jackson2JsonRedisSerializer);
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
}
@Component
public class RedisUtil {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
/** 设置过期时间(秒)*/
public boolean expire(String key, long time) {
if (time > 0) {
return redisTemplate.expire(key, time, TimeUnit.SECONDS);
}
return true;
}
/** 获取过期时间(秒)
* @return 返回0表示永久有效
* */
public long getExpire(String key) {
return redisTemplate.getExpire(key, TimeUnit.SECONDS);
}
/** 判断 key是否存在*/
public boolean hasKey(String key) {
return redisTemplate.hasKey(key);
}
/** 删除缓存*/
@SuppressWarnings("unchecked")
public void del(String... key) {
if (key != null && key.length > 0) {
if (key.length == 1) {
redisTemplate.delete(key[0]);
} else {
redisTemplate.delete(CollectionUtils.arrayToList(key));
}
}
}
/**
* String(字符串)
* */
/** 获取指定缓存*/
public Object get(String key) {
return key == null ? null : redisTemplate.opsForValue().get(key);
}
/** 设置缓存*/
public void set(String key, Object value) {
redisTemplate.opsForValue().set(key, value);
}
/** 设置缓存, 附加过期时间(秒)*/
public void set(String key, Object value, long time) {
if (time > 0) {
redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
} else {
set(key, value);
}
}
/**
* Hash(哈希)
* */
/** 获取 hash里面指定字段的值*/
public Object hmget(String key, String field) {
return redisTemplate.opsForHash().get(key, field);
}
/** 从 hash中读取全部的域和值*/
public Map<Object, Object> hgetall(String key) {
return redisTemplate.opsForHash().entries(key);
}
/** 设置一个 hash字段值, 如不存在将会创建*/
public void hset(String key, String field, Object value) {
redisTemplate.opsForHash().put(key, field, value);
}
/** 设置一个 hash字段值, 如不存在将会创建, 附加过期时间(秒)*/
public void hset(String key, String field, Object value, long time) {
redisTemplate.opsForHash().put(key, field, value);
if (time > 0) {
expire(key, time);
}
}
/** 设置多个 hash字段值*/
public void hmset(String key, Map<String, Object> map) {
redisTemplate.opsForHash().putAll(key, map);
}
/** 设置多个 hash字段值, 附加过期时间(秒)*/
public void hmset(String key, Map<String, Object> map, long time) {
redisTemplate.opsForHash().putAll(key, map);
if (time > 0) {
expire(key, time);
}
}
/** 删除一个或多个 hash的 field*/
public void hdel(String key, Object... field) {
redisTemplate.opsForHash().delete(key, field);
}
/** 判断 field是否存在于 hash中*/
public boolean hexists(String key, String field) {
return redisTemplate.opsForHash().hasKey(key, field);
}
/**
* List(列表)
* */
/** 从列表中获取指定返回的元素*/
public List<Object> lrange(String key, long start, long end) {
return redisTemplate.opsForList().range(key, start, end);
}
/** 获取 key对应的 list的长度*/
public long llen(String key) {
return redisTemplate.opsForList().size(key);
}
/** 获取一个元素, 通过其索引列表*/
public Object lindex(String key, long index) {
return redisTemplate.opsForList().index(key, index);
}
/** 从队列的右边入队一个元素*/
public long rpush(String key, Object value) {
return redisTemplate.opsForList().rightPush(key, value);
}
/** 从队列的右边入队一个元素, 附加过期时间(秒)*/
public long rpush(String key, Object value, long time) {
Long count = redisTemplate.opsForList().rightPush(key, value);
if (time > 0)
expire(key, time);
return count;
}
/** 从队列的右边入队多个元素*/
public long rpush(String key, List<Object> value) {
return redisTemplate.opsForList().rightPushAll(key, value);
}
/** 从队列的右边入队多个元素, 附加过期时间(秒)*/
public long rpush(String key, List<Object> value, long time) {
Long count = redisTemplate.opsForList().rightPushAll(key, value);
if (time > 0)
expire(key, time);
return count;
}
/** 从队列的右边出队一个元*/
public Object rpop(String key) {
return redisTemplate.opsForList().rightPop(key);
}
/** 从队列的左边入队一个元素*/
public long lpush(String key, Object value) {
return redisTemplate.opsForList().leftPush(key, value);
}
/** 从队列的左边入队一个元素, 附加过期时间(秒)*/
public long lpush(String key, Object value, long time) {
Long count = redisTemplate.opsForList().leftPush(key, value);
if (time > 0)
expire(key, time);
return count;
}
/** 从队列的左边入队多个元素*/
public long lpush(String key, List<Object> value) {
return redisTemplate.opsForList().leftPushAll(key, value);
}
/** 从队列的左边入队多个元素, 附加过期时间(秒)*/
public long lpush(String key, List<Object> value, long time) {
Long count = redisTemplate.opsForList().leftPushAll(key, value);
if (time > 0)
expire(key, time);
return count;
}
/** 从队列的左边出队一个元素*/
public Object lpop(String key) {
return redisTemplate.opsForList().leftPop(key);
}
/**
* Sorted Set(有序集合)
* */
/** 添加到有序set的一个成员, 或更新的分数, 如果它已经存在*/
public boolean zadd(String key, Object value, double scoure) {
return redisTemplate.opsForZSet().add(key, value, scoure);
}
/** 添加到有序set的一个成员, 或更新的分数, 如果它已经存在, 附加过期时间(秒)*/
public boolean zadd(String key, Object value, double scoure, long time) {
final Boolean flag = redisTemplate.opsForZSet().add(key, value, scoure);
if (time > 0)
expire(key, time);
return flag;
}
/** 获取一个排序的集合中的成员数量*/
public long zcard(String key) {
return redisTemplate.opsForZSet().zCard(key);
}
/** 根据指定的index返回，返回sorted set的成员列表*/
public Set<Object> zrange(String key, long start, long stop) {
return redisTemplate.opsForZSet().range(key, start, stop);
}
/** 从排序的集合中删除一个或多个成员*/
public long zrem(String key, Object...value) {
return redisTemplate.opsForZSet().remove(key, value);
}
}
@Autowired
private RedisUtil redisUtil;
redisUtil.set("string", "普通字符串值!");
log.info("string: {}", redisUtil.get("string"));
如果您觉得有帮助，欢迎点赞哦 ~ 谢谢!!
