Redis学习总结（中）——事务持久化和主从复制
Posted AC_Jobim
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Redis学习总结（中）——事务持久化和主从复制相关的知识，希望对你有一定的参考价值。
Redis学习总结（中）——事务、持久化和主从复制
一、Redis的事务操作
- redis事务就是一个命令执行的队列，将一系列预定义命令包装成一个整体（一个队列）。当执行时，一次性按照添加顺序依次执行，中间不会被打断或者干扰
1.1 事务的操作和错误处理
事务的操作multi、exec、discard：
-
开启事务
multi
设定事务的开启位置，此指令执行后，后续的所有指令均加入到事务中 -
执行事务
exec
设定事务的结束位置，同时执行事务。与multi成对出现，成对使用
注意：加入事务的命令暂时进入到任务队列中，并没有立即执行，只有执行exec命令才开始执行 -
取消事务
discard
终止当前事务的定义，发生在multi之后，exec之前
即：从输入Multi命令开始，输入的命令都会依次进入命令队列中，但不会执行，直到输入Exec后，Redis会将之前的命令队列中的命令依次执行。组队的过程中可以通过discard来放弃组队。
事务的错误处理：
-
组队中某个命令出现了报告错误，执行时整个的所有队列都会被取消。
-
如果执行阶段某个命令报出了错误，则只有报错的命令不会被执行，而其他的命令都会执行，不会回滚。
1.2 Watch锁
watch锁是一种乐观锁的概念：
乐观锁(Optimistic Lock), 顾名思义，就是很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号等机制。乐观锁适用于多读的应用类型，这样可以提高吞吐量。Redis就是利用这种check-and-set机制实现事务的。
-
watch
在执行multi之前，先执行watch key1 [key2],可以监视一个(或多个) key ，如果在执行exec前如果key被别的线程操作了，则终止事务执行。 -
unwatch
取消 WATCH 命令对所有 key 的监视。
如果在执行 WATCH 命令之后，EXEC 命令或DISCARD 命令先被执行了的话，那么就不需要再执行UNWATCH 了。
1.3 Redis_事务_秒杀案例
Redis中记录商品的库存数量和秒杀成功者清单
1.3.1 使用事务（解决超卖）+连接池（解决超时问题）
存在的问题：
- 问题一：超卖问题
使用
事务（乐观锁）解决 - 问题二：链接超时问题
解决：使用
连接池
jedis连接资源的创建与销毁是很消耗程序性能，所以jedis为我们提供了
jedis的池化技术，jedisPool在创建时初始化一些连接资源存储到连接池中，使用jedis连接资源时不需要创建，而是从连接池中获取一个资源进行redis的操作，使用完毕后，不需要销毁该jedis连接资源，而是将该资源归还给连接池，供其他请求使用。
常用参数:
MaxTotal：控制一个pool可分配多少个jedis实例，通过pool.getResource()来获取；如果赋值为-1，则表示不限制；如果pool已经分配了MaxTotal个jedis实例，则此时pool的状态为exhausted。maxIdle：控制一个pool最多有多少个状态为idle(空闲)的jedis实例；MaxWaitMillis：表示当borrow一个jedis实例时，最大的等待毫秒数，如果超过等待时间，则直接抛JedisConnectionException；testOnBorrow：获得一个jedis实例的时候是否检查连接可用性（ping()）；如果为true，则得到的jedis实例均是可用的；
代码示例
Redis连接池代码：
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.2.4", 6379, 60000 );
}
}
}
return jedisPool;
}
public static void release(JedisPool jedisPool, Jedis jedis) {
if (null != jedis) {
jedisPool.returnResource(jedis);
}
}
}
Servlet代码：
public class SecKillServlet extends HttpServlet {
private static final long serialVersionUID = 1L;
public SecKillServlet() {
super();
}
@Override
protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
String userid = new Random().nextInt(50000) +"" ;
String prodid = req.getParameter("prodid");
boolean isSuccess=SecKill_redis.doSecKill(userid,prodid);
// boolean isSuccess= SecKill_redisByScript.doSecKill(userid,prodid);
resp.getWriter().print(isSuccess);
}
}
Redis操作代码：
public class SecKill_redis {
//秒杀过程
public static boolean doSecKill(String uid,String prodid) throws IOException {
//1 uid和prodid非空判断
if(uid == null || prodid == null) {
return false;
}
//2 连接redis
//通过连接池得到jedis对象
JedisPool jedisPool = JedisPoolUtil.getJedisPoolInstance();
Jedis jedis = jedisPool.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) < 1) {
System.out.println("秒杀已经结束了");
jedis.close();
return false;
}
//7 秒杀过程
//使用事务
Transaction multi = jedis.multi();
//组队操作
//7.1 库存-1
multi.decr(kcKey);
//7.2 把秒杀成功用户添加清单里面
multi.sadd(userKey,uid);
//执行
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;
}
}
前端发起请求
<%@ 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>
1.3.2 使用LUA脚本解决库存依赖问题
问题：已经秒光，可是还有库存。原因，就是乐观锁导致很多请求都失败。先点的没秒到，后点的可能秒到了。
解决：
将复杂的或者多步的redis操作，写为一 个脚本，一次提交给redis执行，减少反复连接redis的次数。提升性能。
LUA脚本是类似redis事务，有一定的原子性，不会被其他命令插队，可以完成一些redis事务性的操作。
但是注意redis的lua脚本功能，只有在Redis 2.6以上的版本才可以使用。
通过lua脚本解决争抢问题，实际上是redis 利用其单线程的特性，用任务队列的方式解决多任务并发问题。
LUA脚本
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;
Redis使用LUA脚本代码
public class SecKill_redisByScript {
private static final org.slf4j.Logger logger =LoggerFactory.getLogger(SecKill_redisByScript.class) ;
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();
Jedis jedis = new Jedis("192.168.2.4", 6379);
//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;
}
}
1.3.3 使用工具ab来模拟并发
CentOS6 默认安装
CentOS7需要手动安装
安装ab工具：yum install httpd-tools
执行代码：
ab -n 2000 -c 200 -k -p ~/postfile -T application/x-www-form-urlencoded http://192.168.81.1:8080/seckill/doseckill
-n 连接数 -c 并发数
二、Redis持久化
Redis的高性能是由于其将所有数据都存储在了内存中，为了使Redis在重启之后仍能保证数据不丢失，需要将数据从内存中同步到硬盘中，这一过程就是持久化。Redis支持两种方式的持久化，一种是 RDB方式，一种是 AOF方式。可以单独使用其中一种或将二者结合使用。
RDB持久化（默认支持，无需配置）
该机制是指在指定的时间间隔内将内存中的数据集快照写入磁盘。快照(Snapshot)也称为RDB持久化方式AOF持久化
该机制将以日志的形式记录服务器所处理的每一个写操作，在Redis服务器启动之初会读取该文件来重新构建redis数据库，以保证启动后数据库中的数据是完整的。- 无持久化
我们可以通过配置的方式禁用Redis服务器的持久化功能，这样我们就可以将Redis视为一个功能加强版的memcached了。 redis可以同时使用RDB和AOF
2.1 RDB持久化方式
2.1.1 RDB持久化特点
是什么？
在指定的时间间隔内将内存中的数据集快照写入磁盘， 也就是行话讲的Snapshot快照，它恢复时是将快照文件直接读到内存里
备份是如何执行的？
Redis会单独创建（fork）一个子进程来进行持久化，会先将数据写入到 一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。 整个过程中，主进程是不进行任何IO操作的，这就确保了极高的性能 如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。RDB的缺点是最后一次持久化后的数据可能丢失。
Fork？
-
Fork的作用是复制一个与当前进程一样的进程。新进程的所有数据（变量、环境变量、程序计数器等） 数值都和原进程一致，但是是一个全新的进程，并作为原进程的子进程
-
在Linux程序中，fork()会产生一个和父进程完全相同的子进程，但子进程在此后多会exec系统调用，出于效率考虑，Linux中引入了“写时复制技术”
-
一般情况父进程和子进程会共用同一段物理内存，只有进程空间的各段的内容要发生变化时，才会将父进程的内容复制一份给子进程。
2.1.2 快照生成方式
- 客户端方式:
BGSAVE和SAVE指令 - 服务器配置自动触发
2.1.2.1 save或者bgsave命令
bgsave命令：
- 客户端可以使用
BGSAVE命令来创建一个快照,当redis服务器接收到客户端的BGSAVE命令时,redis会调用fork来创建一个子进程,然后子进程负责将快照写入磁盘中,而父进程则继续处理命令请求。即：Redis会在后台异步进行快照操作， 快照同时还可以响应客户端请求。
save命令：
- 客户端还可以使用
SAVE命令来创建一个快照,接收到SAVE命令的redis服务器在快照创建完毕之前将不再响应任何其他的命令。即：使用SAVE命令在快照创建完毕之前,redis处于阻塞状态,无法对外服务(写操作)
2.1.2.2 自动触发（重点）
-
如果用户在
redis.conf中设置了save配置选项,redis会在save选项条件满足之后自动触发一次BGSAVE命令, 如果设置多个save配置选项,当任意一个save配置选项条件满足,redis也会触发一次BGSAVE命令
-
表示
900S(15分钟), key发生1次变化, 就触发一次 bgsave命令, 持久化一次 -
表示
300S(5分钟), key发生10次变化, 就触发一次bgsave命令, 持久化一次 -
表示
60S(1分钟), key发生10000次变化, 就触发一次bgsave命令, 持久化一次
上面自动触发的规则: 标明key改变的越频繁, 触发快照持久化到硬盘的时间就越短;
2.1.2.3 服务器接收客户端shutdown指令
- 当
redis服务器接收到redis客户端发来的shutdown指令关闭服务器时,会执行一个save命令,阻塞所有的客户端,不再执行客户端执行发送的任何命令,并且在save命令执行完毕之后关闭服务器
2.1.3 RDB相关配置
-
rdb文件名
在redis.conf中配置文件名称，默认为dump.rdb
-
配置文件位置
rdb文件的保存路径，也可以修改。默认为Redis启动时命令行所在的目录下
dir “/myredis/”
-
stop-writes-on-bgsave-error
后台存储过程中如果出现错误现象，是否停止保存操作。推荐yes. -
rdbcompression 压缩文件
对于存储到磁盘中的快照，可以设置是否进行压缩存储。如果是的话，redis会采用LZF算法进行压缩。如果你不想消耗CPU来进行压缩的话，可以设置为关闭此功能，但会使存储的文件变大（巨大）。推荐yes. -
rdbchecksum 检查完整性
在存储快照后，还可以让redis使用CRC64算法来进行数据校验，但是这样做会增加大约10%的性能消耗，如果希望获取到最大的性能提升，可以关闭此功能。推荐yes.
2.1.4 RDB的备份
- 将*.rdb的文件拷贝到别的地方
- 关闭Redis
- 先把备份的文件拷贝到工作目录下
cp dump2.rdb dump.rdb - 启动Redis, 备份数据会直接加载
2.1.5 RDB持久化的优缺点:
优点：
- RDB是一个紧凑压缩的二进制文件，存储效率较高
- RDB内部存储的是redis在某个时间点的数据快照，非常适合用于数据备份，全量复制等场景
- RDB恢复数据的速度要比AOF快很多
应用：服务器中每X小时执行bgsave备份，并将RDB文件拷贝到远程机器中，用于灾难恢复
缺点：
- RDB方式无论是执行指令还是利用配置，无法做到实时持久化，具有较大的可能性丢失数据
- bgsave指令每次运行要执行fork操作创建子进程，要牺牲掉一些性能
- 基于快照思想，每次读写都是全部数据，当数据量巨大时，效率非常低
2.2 Redis持久化之AOF
2.2.1 AOF持久化的特点
AOF持久化可以将所以上是关于Redis学习总结（中）——事务持久化和主从复制的主要内容，如果未能解决你的问题，请参考以下文章
Redis学习总结（21）——Redis持久化是如何做的？RDB和AOF对比分析