文章目录

• 1. Redis 事务
• • 1.1 Redis事务的作用
  • 1.2 Multi ， Exec ， discard 命令(事务常用的三个命令)
  • 1.3 事务的错误处理
• 2. Redis 事务冲突的问题
• • 2.1 什么是事务冲突？
  • 2.2 如何解决事务冲突？ 乐观锁和悲观锁
  • 2.3 watch 命令 和 unwatch命令
  • 2.4 Redis 事务三个特性
• 3. Redis 事务 秒杀案例
• • 3.1 秒杀案例的 基本实现
  • 3.2 使用ab 来模拟并发秒杀
  • 3.3 通过连接池解决 连接超时的问题
  • 3.4 如何解决并发出现的超卖现象(库存为负数)
  • 3.5 库存的 遗留问题
  • 3.6 LUA脚本在Redis中的优势
• 4. Redis 持久化之 RDB(Redis DataBase)
• • 4.1 持久化
  • 4.2 RDB 持久化(备份)是如何进行的？
  • 4.3 Fork
  • 4.4 redis配置文件 配置RDB相关信息
  • 4.5 RDB 优劣比较
  • 4.6 rdb的 备份和恢复
• 5. Redis 持久化之 AOF(Append Only File)
• • 5.1 什么是AOF?
  • 5.2 AOF的 开启
  • 5.3 AOF 备份和恢复
  • 5.4AOF 同步频率设置
  • 5.5 AOF rewrite重写操作(压缩)
  • 5.6 AOF执行流程 和 优缺点
• 6. Redis 主从复制
• 7. Redis 主从复制 搭建一主多从
• 8. Redis 主从复制的原理
• 9. Redis 主从复制之 一主二仆
• • 9.1 当 从服务器 宕机了
  • 9.2 当 主服务器 宕机了
• 10. Redis 主从复制之 薪火相传 ， 反客为主
• • 10.1 薪火相传
  • 10.2 反客为主
• 11. Redis 主从复制之 哨兵模式(sentinel)
• • 11.1 什么是哨兵模式？
  • 11.2 开启哨兵模式
  • 11.3 哨兵模式的效果
  • 11.4 当主机宕机后 根据什么从 从机 中选举 主机 呢？
  • 11.5 复制延时的弊端
  • 11.6 哨兵模式下，Java代码如何指定 主服务器？

1. Redis 事务

1.1 Redis事务的作用

Redis 事务是一个单独的隔离操作，事务中的所有命令都会序列化，按顺序地执行(因为，单线程+IO多路复用)。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。

Redis 事务的主要作用就是串联多个命令防止别的命令插队。

1.2 Multi ， Exec ， discard 命令(事务常用的三个命令)

multi 和 exec 命令执行效果：

而discard命令是放弃组队，就类似mysql的回滚。

1.3 事务的错误处理

在组队阶段，如果我们某个命令出现了错误，整个队列中的命令都会被取消。

下面这段英文要知道！！

如果执行阶段某个命令报出了错误，则只有报错的命令不会被执行，而其他正常的命令都会执行，不会回滚。

2. Redis 事务冲突的问题

2.1 什么是事务冲突？

像下面这种多个事务，同时对一个数据进行操作的情况，就是事务冲突。

2.2 如何解决事务冲突？ 乐观锁和悲观锁

解决事务冲突就必须要用到锁。

锁一般分为悲观锁和乐观锁。

悲观锁：无论对数据进行什么操作的时候，都先会上锁，这样别人就不能对该数据进行增删改查操作了。除非解锁。

(这个锁的缺点就是效率很低。)

乐观锁：与悲观锁相反。但是版本号的机制要注意！！

版本号机制要知道！！！

2.3 watch 命令 和 unwatch命令

通过watch命令来监视一个或多个键。

unwatch命令就是取消watch命令对所有key的监视。

如果在执行watch命令之后，执行了exec命令或discard命令的话，那就不需要再执行unwatch了。

2.4 Redis 事务三个特性

首先，redis是不支持ACID四特性的。redis对于事务有专门自己的三个特性。

3. Redis 事务 秒杀案例

3.1 秒杀案例的 基本实现

首先，确定库存和成功用户的要使用redis的数据类型，如下：

package com.itholmes.jedis;

import java.util.Random;

import redis.clients.jedis.Jedis;

public class SecKill_redis {
	public static void main(String[] args) {
		
		Random random = new Random();
		
		//模拟随机用户
		String uid = "";
		for(int i=0;i<4;i++) {
			//随机生成10以内的一个数字
			int rand = random.nextInt(10);
			uid += rand;
		}
		
		//商品标签为0101
		String prodid = "0101";
		
		//假设多个用户调用这个方法进行秒杀，那么数据库中的对应key-value就行操作的！
		doSecKill(uid, prodid);
	}
	
	//秒杀过程,uid就是用户id，prodid就是库存id
	public static boolean doSecKill(String uid,String prodid) {
		
		//1.uid和prodid非空判断
		if(uid == null || prodid == null) {
			return false;
		}
		
		//2.连接redis
		Jedis jedis = new Jedis("39.103.163.156",6379);
		
		//3.拼接key
		//3.1 库存key
		String kuncunKey = "sk:"+prodid+":qt";
		//3.2 秒杀成功用户key
		String userKey = "sk:"+prodid+":user";
		
		//4.获取库存，如果库存本身等于null，表示秒杀还没有开始
		String kuncun = jedis.get(kuncunKey);
		if(kuncun == null) {
			System.out.println("秒杀还没有开始，请等待！");
			jedis.close();
			return false;
		}
		
		//5.判断用户是否重复秒杀操作(每个用户秒杀只能秒杀一次)
		Boolean sismember = jedis.sismember(userKey, uid);
		if(sismember) {
			System.out.println("已经秒杀成功了，不能重复秒杀！");
			jedis.close();
			return false;
		}
		
		//6.判断如果商品数量，库存数量 小于1了，说明秒杀结束了
		int i = Integer.parseInt(kuncun);
		if(i<1) {
			System.out.println("秒杀已经结束了");
			jedis.close();
			return false;
		}
		
		
		//7. 秒杀过程
		//7.1 库存要减1
		jedis.decr(kuncunKey);
		//7.2 把秒杀成功用户添加到清单里面
		jedis.sadd(userKey, uid);
		System.out.println("秒杀成功了");
		
		jedis.close();
		return true;
	}
}

3.2 使用ab 来模拟并发秒杀

yum install httpd-tools 或 apt-get install httpd-tools 安装httpd-tools工具。

我们要使用它的ab，ab是apache性能测试工具。

ab 的参数。

• -n 表示当前请求次数。
• -c 表示当前的并发次数。
• -p 表示post方法发送过来的参数存放位置。
• -T 表示post方法设置类型。

通过这个方式来测试并发效果很好！！

就3.1的秒杀代码而言，有严重的并发问题！例如：库存成为了负数(超卖问题)，再者就是太多的请求，会有超时问题！

3.3 通过连接池解决 连接超时的问题

redis不能同时处理多个请求，就有了连接超时的问题。

连接池：节省每次连接redis服务带来的消耗，把连接好的实例反复利用。

模拟一个jedis数据库池的效果：

package com.itholmes.jedis;

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);
					
					jedisPool = new JedisPool(poolConfig,"39.103.163.156",6379,60000);
				}
			}
		}
		return jedisPool;
	}
	
	
	public static void release(JedisPool jedisPool, Jedis jedis) {
		if(null != jedis) {
			//关闭归还连接。
			jedis.close();
		}
	}
}

3.4 如何解决并发出现的超卖现象(库存为负数)

首先，可以通过watch乐观锁来解决！（记住乐观锁的版本号对应机制）

注意下面是如何开启redis事务(组或队列)的！！

package com.itholmes.jedis;

import java.util.List;
import java.util.Random;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Transaction;

public class SecKill_redis {
	public static void main(String[] args) {
		
		Random random = new Random();
		
		//模拟随机用户
		String uid = "";
		for(int i=0;i<4;i++) {
			//随机生成10以内的一个数字
			int rand = random.nextInt(10);
			uid += rand;
		}
		
		//商品标签为0101
		String prodid = "0101";
		
		//假设多个用户调用这个方法进行秒杀，那么数据库中的对应key-value就行操作的！
		doSecKill(uid, prodid);
	}
	
	//秒杀过程,uid就是用户id，prodid就是库存id
	public static boolean doSecKill(String uid,String prodid) {
		
		//1.uid和prodid非空判断
		if(uid == null || prodid == null) {
			return false;
		}
		
		//2.连接redis
		//Jedis jedis = new Jedis("39.103.163.156",6379);
		
		//这样我们就可以通过连接池得到jedis对象,通过连接池解决超时问题。
		JedisPool jedisPool = JedisPoolUtil.getJedisPoolInstance();
		Jedis jedis = jedisPool.getResource();
		
		//3.拼接key
		//3.1 库存key
		String kuncunKey = "sk:"+prodid+":qt";
		//3.2 秒杀成功用户key
		String userKey = "sk:"+prodid+":user";
		
		
		//监视库存,相当于开启了乐观锁(记住乐观锁的版本号对应机制)
		jedis.watch(kuncunKey);
		
		//4.获取库存，如果库存本身等于null，表示秒杀还没有开始
		String kuncun = jedis.get(kuncunKey);
		if(kuncun == null) {
			System.out.println("秒杀还没有开始，请等待！");
			jedis.close();
			return false;
		}
		
		//5.判断用户是否重复秒杀操作(每个用户秒杀只能秒杀一次)
		Boolean sismember = jedis.sismember(userKey, uid);
		if(sismember) {
			System.out.println("已经秒杀成功了，不能重复秒杀！");
			jedis.close();
			return false;
		}
		
		//6.判断如果商品数量，库存数量 小于1了，说明秒杀结束了
		int i = Integer.parseInt(kuncun);
		if(i<1) {
			System.out.println("秒杀已经结束了");
			jedis.close();
			return false;
		}
		
		
		//7. 秒杀过程
		
		//开始一个队列(组或者事务，都可以理解)
		Transaction multi = jedis.multi();
		
		//组队操作
		//7.1 库存要减1
		multi.decr(kuncunKey);
		//7.2 把秒杀成功用户添加到清单里面
		multi.sadd(userKey, uid);
		
		//开启事务后，就使用事务对应的方法！！来进行组队等等操作。就不单独来操作。
		//jedis.decr(kuncunKey);
		//jedis.sadd(userKey, uid);
		
		//执行队列命令。exec()方法会给我们返回一个list集合结果内容。
		List<Object> results = multi.exec();
		
		if(results == null || results.size() == 0) {
			System.out.println("秒杀失败！！");
			jedis.close();
			return false;
		}
		
		System.out.println("秒杀成功了");
		
		jedis.close();
		return true;
	}
}

3.5 库存的 遗留问题

乐观锁会造成库存遗留的问题！

解决办法，最先想到的是悲观锁，但是悲观锁在redis中，不行！！
redis中默认不能直接使用悲观锁(不支持悲观锁)，只能使用乐观锁。

• 原因：Redis 作为缓存服务器使用时，以读操作为主，很少写操作，相应的操作被打断的几率较少。不采用悲观锁是为了防止降低性能。

用嵌入式脚本语言Lua来，来解决库存遗留问题。

3.6 LUA脚本在Redis中的优势

就是将复杂的或者多步的redis操作，写为一个LUA脚本，一次提交给redis执行。

下面就是一个LUA脚本(了解)：

4. Redis 持久化之 RDB(Redis DataBase)

4.1 持久化

什么是持久化？

• 就是将数据写到硬盘中，能长时间存储！这个过程就叫做持久化。

Redis有两种不同形式的持久化方式：

• RDB(Redis DataBase)，redis默认持久化方式。
• AOF(Append Of File)

4.2 RDB 持久化(备份)是如何进行的？

什么是RDB？

• 在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是我们将Snapshot快照，它恢复时是将快照文件在直接读到内存里。这种方式就叫做RDB。

RDB执行过程：

RDB创建fork，以及RDB的缺点(最后一次持久化后的数据可能丢失) 要牢记。

4.3 Fork

RDB有一个非常重要的底层原理叫做写时复制技术。

fork的介绍：

4.4 redis配置文件 配置RDB相关信息

首先，我们找到redis.conf中的SNAPSHOTTING，这一块就是配置快照相关的信息。

配置dbfilename：配置持久化后的文件名称。

配置dir：默认是再当前启动redis服务的命令下生成持久化文件。

配置stop-writes-on-bgsave-error：当redis无法写入磁盘的话，直接关掉Redis的写操作。推荐yes。

配置 rdbcompression：压缩文件，推荐yes，压缩么节约空间。

配置rdbchecksum：检查完整性，就是一个数据效验的功能。

配置save [秒钟] [写操作次数]：就是在多少秒内，有多少写操作，就进行持久化操作。

解释一下上面英文内容内容：

• 在1个小时内，有1个key发生了变化，就进行持久化操作同步。
• 在5分钟内，有100个key发生了变化，就进行持久化操作同步。
• 在60秒内，有10000个key发生了变化，就进行持久化操作同步。
  
  

对应的两个持久化的redis命令： save 和 bgsave 命令
(在加上一个lastsave命令。)

4.5 RDB 优劣比较

优点：周期性的进行持久化操作。

缺点：最后一次的快照信息可能丢失！

4.6 rdb的 备份和恢复

如果我们存储的dump.rdb出现了问题！我们就可以通过备份的文件来恢复。

下面的cp一份持久化的数据就是redis备份操作。将cp的数据改个成redis默认名字放到相同目录，重启redis服务就是redis恢复操作。

首先，我们cp一份dump.rdb数据，这样dump.rdb出现问题的时候，我们只需要将cp的备份数据，改个名字，放过来就好了。这就算是一个备份效果。

5. Redis 持久化之 AOF(Append Only File)

5.1 什么是AOF?

Append Only File的意思可以理解为仅仅向文件中追加信息。

AOF持久化的原理：

5.2 AOF的 开启

AOF持久化默认是不开启的。

配置AOF的相关信息，就在append only node模块下。

配置appendonly ： 配置为yes表示开启AOF模式。

配置appendfilename：持久化后生成的文件名字。

当我们配置上面的信息，重启redis服务后，就开启了AOF持久化！

注意一个问题：

• 其实这个时间段AOF和RDB是同时开启的，我们并没有在配置文件关闭RDB之类的操作。当AOF和RDB同时开启，系统默认取AOF的数据（数据不会存在丢失）

当我们开启了AOF，进行一个写操作就往aof文件中追加一个写操作命令。

5.3 AOF 备份和恢复

AOF 备份和恢复的效果和rdb一样的：

AOF多了一个异常恢复的效果。

异常恢复：

• 通过/usr/local/bin/redis-check-aof --fix 可以将坏的aof文件恢复成为正常的。
  

例如：我们将appendonly.aof，随便改了改。

这样我们在进行redis-cli连接时，就会拒接连接。

然后，我们就可以通过rdis-check-aof --fix xxx.aof命令来修复aof文件了。

5.4AOF 同步频率设置

配置appendfsync：同步频率设置。

5.5 AOF rewrite重写操作(压缩)

rewrite重写 压缩操作：

说的简单一点就是 多个指令给他重写成为一个执行进行操作。

什么时候出发重写操作？

AOF 重写对应两个相关配置：

案例：

5.6 AOF执行流程 和 优缺点

AOF执行流程：

AOF的优点：

AOF的缺点：

6. Redis 主从复制

主机数据更新后根据配置和策越，自动同步到备份机的master/slaver机制。

其中，master以写为主，slave以读为主。

就像下面这张图，这就是主从复制，读写分离。

主从复制的优点：

• 实现读写分离，性能扩展。
• 容灾快速恢复：就是当我们在 从数据库 进行读操作，如果这个 从数据库 出现问题，宕机了，就快速切换到另一个从数据库 进行读操作。这个操作就叫做容灾快速恢复。

容灾快速恢复是用来解决从数据库发生问题宕机后，切换另一个台从数据库。那么如果主数据库出现问题怎么办呢？

那么对应解决主数据库 如果出现问题宕机了该怎么办？就用到了Redis集群。

虽说，每个主从复制要求是一主多从的效果，就是一台主数据库，多台从数据库。而Redis集群就把多个一主多从 进行了联系起来。当用户访问的这台主数据库出现问题，就通过集群切换到另一个个一主多从的主数据库 中。

7. Redis 主从复制 搭建一主多从

这里实现一个一主两从的效果。

第一步：创建一个文件夹用来放置三个redis数据库的配置文件，持久化文件等等。这里我创建的是myredis文件夹。

第二步：复制cp我们的redis.conf配置文件到文件夹中，作为一个公共配置文件部分（配置文件可以去安装redis的目录下获取）。

第三步：配置一主两从，创建三个配置文件。

三个配置文件修改的信息分别是：

• include导入redis.conf的公共部分。
• 修改pidfile的路径和名称。
• 修改port端口
• 修改dbfilename持久化文件名称。

include的作用是引入公共部分。这里我们引入了之前copy的redis.conf配置文件。就是分别对三个redis数据库进行配置。

(一般我们命名就是在后面添加上对应的数字就行，我这里用的6379，6380，6381作为三个redis服务器，整好设置不同的端口。)

第四步：分别创建并配置好三个配置文件。

第五步：redis-server [对应的.conf文件] 启动三台redis服务器。并且使用slaveof命令来配置从服务器。

如果我们想要连接其中一台redis服务器的话，就使用redis-cli -p [端口号] 就可以了。通过端口号来打开redis服务器的客户端。

我们可以使用info replication 命令来查看主从复制的相关信息。

目前，三个redis服务器都是 主服务器。接下来就要配置从服务器 例如：将6379作为主服务器，6380和6381作为从服务器。

配从(库) 不配主(库)
在6380 和 6381 上执行：slaveof [ip] [port] 。 来进行配置。

6380从服务器：

6381从服务器：

6379主服务器：

这样一主两从的效果就做出来了。

第六步：测试三台redis服务器的效果。

• 测试1：往主服务器进行写操作，两台从服务器也能读取到。
  
• 测试2：只有主服务器能进行写操作，其他两个从服务器不能进行写操作，只能进行读操作(读写分离)。
  

8. Redis 主从复制的原理

9. Redis 主从复制之 一主二仆

9.1 当 从服务器 宕机了

如果 从服务器 发生了问题宕机了，那么当该 从服务器 重启启动起来后，它变成了一台 主服务器(info replication命令，显示master)。如果想让它重新变成从服务器还是需要slaveof [ip] [port] 命令来指定。

当然，执行了slaveof命令后，该服务器会将主服务器的内容复制过来（正常的复制操作）。

9.2 当 主服务器 宕机了

当 主服务器 宕机了出问题了，那么它下面的 从服务器 不做任何事情，仍然指向 主服务器 不会发生变化。

当 主服务器 重新启动起来，麾下仍然有两个 从服务器 ，可以用info replication查看。

10. Redis 主从复制之 薪火相传 ， 反客为主

10.1 薪火相传

什么是薪火相传？

主从服务器想要实现这样的效果也很简单，就是使用slaveof命令一层层套就可以了，就是一台 从服务器 下面还可以有一个从服务器。

主从服务器宕机的情况如下：(和上面一主两仆一样的。)

10.2 反客为主

当一个master宕机后，后面的slave可以立刻升为master，其后面的slave不用做任何修改。

可以使用slaveof no one命令 将从机变为主机。

反客为主缺点就是需要手动输入命令将当前 从机 变成 主机 。为了解决这个缺点，就引出了哨兵模式。

11. Redis 主从复制之 哨兵模式(sentinel)

11.1 什么是哨兵模式？

哨兵模式就是反客为主的自动版，能够后台监控主机是否故障，如果故障了根据投票数自动将从库转换为主库。

效果图如下：

11.2 开启哨兵模式

第一步：先调成一主二仆(别的可以！)，一主二仆效果更佳明显。

第二步：在我们的myredis目录下，新建sentinel.conf文件，这里的名字不能错！

配置sentinel.conf文件：(加入一个命令就可以)

• mymaster就是为要监控的服务器起的名字。
• 别忘记加上地址和端口号。
• 最后面的 1 代表只要有1个哨兵同意迁移，就进行迁移。设置为2代表只要有两个哨兵同意迁移，就进行迁移。
  
  

第三步：启动哨兵。

执行 redis-sentinel sentinel.conf 命令来启动哨兵模式。

11.3 哨兵模式的效果

当主机挂掉，从机选举中产生新的主机。

首先，我将6379主机服务器进行宕机处理(shutdown)，这是哨兵模式下，会进行调整。

当宕机的主机再次重新启动起来，就变成了从机了，并且指向哨兵选出的主机。

11.4 当主机宕机后 根据什么从 从机 中选举 主机 呢？

当 主机 宕机后，根据什么从 从机 中选举 主机 呢？

首先，根据slave-priority(旧版本)或replica-priority(redis6版本)来判断优先级别。值越小，优先级越高。

如果，优先级相同，就考虑偏移量最大的(这里的偏移量值获得原主机最全的)。

如果，偏移量也相同，那就选择runid最小的从服务。(每个redis实例启动后都会随机生成一个40位的runid)

就是靠这三个条件来操作的。

11.5 复制延时的弊端

11.6 哨兵模式下，Java代码如何指定 主服务器？

代码实现如下：

package com.itholmes.sentinel;

import java.util.HashSet;
import java.util.Set;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisSentinelPool;

public class JedisSentinelCode {
	
	private static JedisSentinelPool jedisSentinelPool = null;
	
	public static Jedis getJedisFromSentinel() {
		if(jedisSentinelPool == null) {
			
			Set<String> sentinelSet = new HashSet<String>();
			//设置好ip和redis哨兵模式的端口号
			sentinelSet.add("39.103.163.156:26379");
			
			JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
			jedisPoolConfig.setMaxTotal(10);//最大可用连接数
			jedisPoolConfig.setMaxIdle(5);//最大闲置连接数
			jedisPoolConfig.setMinIdle(5);//最小闲置连接数
			jedisPoolConfig.setBlockWhenExhausted(true);//连接耗尽是否等待
			jedisPoolConfig.setMaxWaitMillis(2000);//等待事件
			jedisPoolConfig.setTestOnBorrow(true);//取连接的时候进行一下测试  ping
			
			//第一个参数，就是我们哨兵配置的主机名，第二个参数就是一个存着服务器IP和哨兵端口号的set类型 ， 第三个参数jedis池配置对象。
			jedisSentinelPool = new JedisSentinelPool("mymaster", sentinelSet,jedisPoolConfig);
		
		}
		Jedis jedis = jedisSentinelPool.getResource();
		return jedis;
	}
	
	public static void main(String[] args) {
		
		Jedis jedis = getJedisFromSentinel();
		
		//测试是否成功连接
		String ping = jedis.ping();
		System.out.println(ping);
		
		//获取info信息
		String info = jedis.info();
		System.out.println(info);
	}
	
}

说说遇到的几个坑。

• 因为我是云服务器，所以我们主机，从机，哨兵模式端口都要配置出入规则。
• 我们在配置sentinel.conf(哨兵配置文件)时的IP经常写成127.0.0.1，这样我们在Java代码中获取到的ip也就127.0.0.1，这就没法访问到云服务器，因为我们要设置为我们的云网段ip地址才行！