Redis入门手册

前段时间因为自己再鼓捣zk和dubbo玩了下分布式的相关内容,想到本身 redis 也是用來实现分布式锁的一种手段，所以心痒之下，整理回顾了redis的相关内容。

知识点:

1. Redis数据结构及命令介绍
2. Redis实现简易的分布式锁
3. Lua语言的简介和入门
4. Redis中 lua 脚本的运用
5. Redis的持久化策略RDB&AOF
6. Redis集群
7. 关于缓存穿透击穿及失效雪崩的思考和解决方案

学习准备:

1. 虚拟机三台 Centos7.2
2. Redis-3.2.11.tar.gz 安装包一个 具体可以在如下网址中下载 download.redis.io/releases/

声明：以下操作都是基于demo层级的实践，由于深度不够所以如果出现有误的请联系本人共同学习成长。

学习前瞻 redis的优势 存储结构：

1）字符类型

2）散列类型

3）列表类型

4）集合

5）有序集合

redis的功能：

1）可以为每个key设置超时时间；expire

2）可以通过列表类型来实现分布式队列

3）支持发布订阅的消息模式 pub sub模型

简单 1）提供了很多命令与redis进行交互

Redis的应用场景 1）数据缓存（商品数据，新闻，热点）

2）单点登录

3）秒杀，抢购

4）网站访问排名

5）应用模块开发

redis的安装

1. 下载redis安装包 解压tar包

   2）进去到redis目录下中执行make命令来编译

   yum install gcc

   3）通过make test测试编译状态【一路ok就代表没问题】

   如果出现You need tcl 8.5 or newer in order to run the Redis test

   那就yum install tcl

make MALLOC=libc

4）通过make install 完成安装【make install [prefix=/path]完成安装】

我们看看bin下面有些什么命令

启动停止redis 先要复制一份conf文件到redis目录下 cp /data/program/redis-3.2.11/redis.conf ../redis.conf

./redis-server ../redis.conf

修改配置文件使其能够以进程形式start

./redis-cli shutdown 以后台进程的方式启动，修改redis.conf daemonize =yes

连接到redis的命令

./redis-cli -h 127.0.0.1 -p 6379

其他命令说明

Redis-server 启动服务

Redis-cli 访问到redis的控制台

redis-benchmark 性能测试的工具

redis-check-aof aof文件进行检测的工具

redis-check-dump rdb文件检查工具

redis-sentinel sentinel 服务器配置

多数据库支持

默认支持16个数据库；可以理解为一个命名空间

跟关系型数据库不一样的点

1. redis不支持自定义数据库名称
2. 每个数据库不能单独设置授权
3. 每个数据库之间并不是完全隔离的。 可以通过flushall命令清空redis实例面的所有数据库中的数据
   通过 select dbid 去选择不同的数据库命名空间 。 dbid的取值范围默认是0 -15

使用入门：

1)获得一个符合匹配规则的键名列表

keys lulf:allen

keys pattern [? / * /[]]

2）判断一个键是否存在 ， EXISTS key

3）type key 去获得这个key的数据结构类型

各种数据结构的使用

1）字符类型

一个字符类型的key默认存储的最大容量是512M

赋值和取值

SET Key Value

get Key

递增数字

incr key

错误的演示【无法保证原子性】

int value= get key;

value =value +1;

set key value;

key的设计

对象类型:对象id:对象属性:对象子属性

建议对key进行分类，同步在wiki统一管理

短信重发机制：sms:limit:mobile 138。。。。。 expire【超时】

incryby key increment 递增指定的整数 decr key 原子递减 append key value 向指定的key追加字符串 strlen key 获得key对应的value的长度 mget key key.. 同时获得多个key的value【建议使用，减少网络传输】 mset key value key value key value … setnx

2）散列类型

hash key value 不支持数据类型的嵌套

比较适合存储对象

person

age 18

sex 男

name allen

hset key field value

hget key filed

hmset key filed value [filed value …] 一次性设置多个值

hmget key field field … 一次性获得多个值

hgetall key 获得hash的所有信息，包括key和value

hexists key field 判断字段是否存在。 存在返回1. 不存在返回0

hincryby

hsetnx

hdel key field [field …] 删除一个或者多个字段

3）列表类型

list, 可以存储一个有序的字符串列表

LPUSH/RPUSH： 从左边或者右边push数据

LPUSH/RPUSH key value value …

｛17 20 19 18 16｝

llen num 获得列表的长度

lrange key start stop ; 索引可以是负数， -1表示最右边的第一个元素

lrem key count value

lset key index value

LPOP/RPOP : 取数据【相当于弹出】

应用场景：可以用来做分布式消息队列

4）集合

set 跟list 不一样的点。 集合类型不能存在重复的数据。而且是无序的

sadd key member [member ...] 增加数据； 如果value已经存在，则会忽略存在的值，并且返回成功加入的元素的数量

srem key member 删除元素

smembers key 获得所有数据

sdiff key key … 对多个集合执行差集运算

sunion 对多个集合执行并集操作, 同时存在在两个集合里的所有值

5）有序集合

zadd key score member

zrange key start stop [withscores] 去获得元素。 withscores是可以获得元素的分数 如果两个元素的score是相同的话，那么根据(0<9<A<Z<a<z) 方式从小到大 网站访问的前10名。

redis的事务处理

MULTI 去开启事务

EXEC 去执行事务

俩个命令在原子事务中

redis的过期时间

expire key seconds

ttl 获得key的过期时间

redis 的发布订阅 pub/sub 模型

publish channel message

subscribe channel [ …]

codis . twmproxy

注释掉，代表外网可以访问

edis的分布式锁 数据库可以做 activemq

缓存 -redis setnx

Zookeeper

分布式锁本质【用来解决什么问题】

分布式架构是多进程的架构,利用第三方解决方案来解决并发情况下的多进程访问共享资源的问题。

1）资源共享的竞争问题

2）数据的安全性

解决分布式锁：

1）zookeeper 有序节点 ， watcher机制

2）数据库

3）redis setnx

分布式锁的实现

package com.Allen.redis;

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

public class RedisManager {
	
	private static JedisPool jedisPool;
	
	static{
		JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
		jedisPoolConfig.setMaxTotal(20);
		jedisPoolConfig.setMaxIdle(10);
		jedisPool=new JedisPool(jedisPoolConfig,"192.168.48.133",6379);
	}
	
	public static Jedis getJedis() throws Exception{
		if(jedisPool!=null){
			return jedisPool.getResource();
		}
		throw new Exception("Jedis is null");
	}
}

复制代码

package com.Allen.redis;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

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

public class RedisLock {
	// 获取锁
	public String getLock(String key, int timeout) {
		try {
			Jedis jedis = RedisManager.getJedis();
			String value = UUID.randomUUID().toString();
			// 超时时间
			long end = System.currentTimeMillis() + timeout;
			while (System.currentTimeMillis() < end) {// 阻塞
				if (jedis.setnx(key, value) == 1) {
					jedis.expire(key, timeout);//过期时间
					// 锁设置成功
					return value;
				}
				if(jedis.ttl(key)==-1){//检测过期时间
					jedis.expire(key, timeout);
				}
				TimeUnit.SECONDS.sleep(1);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

	// 释放锁
	public boolean releaseLock(String key, String value) {
		try {
			Jedis jedis = RedisManager.getJedis();
			while (true) {
				jedis.watch(key);
				// 判断获取锁的线程和当前redis中的锁是同一个
				if (value.equals(jedis.get(key))) {
					// 获取事务
					Transaction transaction = jedis.multi();
					transaction.del(key);
					List<Object> list = transaction.exec();
					if (list == null) {
						continue;
					}
					return true;
				}
				jedis.unwatch();
				break;
			}

		} catch (Exception e) {
			e.printStackTrace();
		}
		return false;
	}

}

复制代码

如果报错

设置如下：

测试一下：

package com.Allen.redis;

public class Test {
	public static void main(String[] args) {
		RedisLock redisLock = new RedisLock();
		// 获取锁
		String lockID = redisLock.getLock("Allen", 10000);
		System.out.println("第一次获取Allen对象锁---> lockID是 " + lockID);
		String lockID2 = redisLock.getLock("Allen", 10000);
		if (null == lockID2) {
			System.out.println("第二次获取Allen对象锁失败");
		} else {
			System.out.println("第二次获取Allen对象锁---> lockID是 " + lockID2);
		}
	}
}

复制代码

redis性能这块关于多路复用机制

IO多路复用机制

同步阻塞IO

同步非阻塞IO

多路复用

lua语言：

好处：

1）减少网络开销

2）原子操作

3）复用性

tar zxf lua-5.3.0.tar.gz cd lua-5.3.0

yum install readline-devel make linux test make install

Lua的语法学习 lua是动态类型的语言 先谈下lua语言的变量，lua变量分成全局变量和局部变量。 a=1; local b=2;

逻辑表达式 加减乘除 +-*/ 关系运算符 == 等于 ~= 不等于

逻辑运算符 and/or not(a and b)

连接字符串

计算字符串长度

逻辑控制语句

foreach local xx={"aa","bb","cc"} for i,v in ipairs(xx) do print(v) end

函数【全局scope 局部local】 scope function（） return end

写一个lua脚本 进入到redis中 eval 后面的就是luo脚本 eval 脚本内容 keynumber key args。。。

-- 对某个ip的频率进行限制 一分钟访问十次 local num=redis.call('incr',KEYS[1]) if tonumber(num)==1 then redis.call('expire',KEYS[1],ARGV[1]) return 1 elseif tonumber(num)>tonumber(ARGV[2]) then return 0 else return 1 end

luo脚本未执行完，其他操作会存在问题

package com.Allen.redis;

import java.util.ArrayList;
import java.util.List;

import redis.clients.jedis.Jedis;

public class LuoDemo {
	public static void main(String[] args) throws Exception {
		Jedis jedis = RedisManager.getJedis();
		String luostr = "local num=redis.call('incr',KEYS[1])\n" + "if tonumber(num)==1 then\n"
				+ "redis.call('expire',KEYS[1],ARGV[1])\n" + "return 1\n"
				+ "elseif tonumber(num)>tonumber(ARGV[2]) then\n" + "return 0\n" + "else\n" + "return 1\n" + "end\n";
		List<String> KEYS=new ArrayList<String>();
		KEYS.add("ip:limit:192.168.48.133");
		List<String> ARGVS=new ArrayList<String>();
		ARGVS.add("60000");
		ARGVS.add("10");
		Object obj=jedis.eval(luostr,KEYS,ARGVS);
		System.out.println(obj);
	}
}


复制代码

redis持久化机制

提供了俩种持久化策略

RDB

RDB的持久化策略，按照规则定时将内存的数据同步到磁盘

redis在指定的情况下会触发快照

1）自己配置的快照规则

我们首先看下redis.conf这个配置文件去看下系统默认的快照规则

save

当900秒内，被更改的key的数量大于1的时候就执行快照

save 900 1

save 300 10

save 60 10000

2）save或者bgsave

执行内存数据同步到磁盘的操作，这个操作会阻塞客户端请求【save】

后台异步执行快照操作，这个操作不会阻塞客户端请求【bgsave】background

3）执行flushall的时候

清除内存所有数据，只要快照规则不为空，那么redis就会执行快照

4）执行复制的时候

redis集群

快照文件如下：

快照的实现原理：

redis会使用fork函数复制一份当前的进程副本（子进程）

父进程可以继续进行客户端请求，子进程会把内存数据同步到磁盘的临时文件上，所以不会影响到当前应用的使用。

redis的优缺点

缺点：redis可能会存在数据丢失的情况，执行快照和执行下一次快照中间的数据可能会丢失，宕机。

优点：可以最大化redis的性能

AOF AOF的持久化策略，每次执行完命令后会把命令本身存储下来【类似于实时备份】 俩种持久化策略可以使用一种也可以是同时使用，如果同时使用 重启时候会优先使用AOF还原数据。

redis会把每一条命令追加到磁盘文件中，会对性能有所影响。

改成yes 即开启了aof

修改redis.conf 中的appendonly yes

重启执行对数据的变更命令，会在bin目录下生成对应的.aof文件，aof会记录所有的操作命令

如下俩个参数可以对aof文件进行优化

压缩策略

auto-aof-rewrite-percentage 100

ps：表示当前aof文件大小超过上一次aof文件大小的百分之多少的时候会进行重写，如果之前没有重写，以启动时的aof文件大小为准

auto-aof-rewrite-min-size 64mb

ps:限制允许重写最小aof文件大小，也就是文件大小肖宇64mb的时候，不需要进行优化

aof重写的原理：

aof重写的整个过程是安全的

Redis 可以在 AOF 文件体积变得过大时，自动地在后台对 AOF 进行重写： 重写后的新 AOF 文件包含了恢复当前数据集所需的最小命令集合。 整个重写操作是绝对安全的，因为 Redis 在创建新 AOF 文件的过程中，会继续将命令追加到现有的 AOF 文件里面，即使重写过程中发生停机，现有的 AOF 文件也不会丢失。 而一旦新 AOF 文件创建完毕，Redis 就会从旧 AOF 文件切换到新 AOF 文件，并开始对新 AOF 文件进行追加操作。AOF 文件有序地保存了对数据库执行的所有写入操作， 这些写入操作以 Redis 协议的格式保存， 因此 AOF 文件的内容非常容易被人读懂， 对文件进行分析（parse）也很轻松

同步磁盘数据

redis每次更改数据的时候，aof机制都会将命令记录到aof文件，但是实际上由于操作系统的缓存机制，数据没有实时地写入硬盘，而是进入硬盘缓存，再通过硬盘缓存机制去刷新保存到文件。【理论上可能出现数据丢失】

appendfsync always 每次执行写入都会同步，最安全，效率最低

appendfsync everysec 每一秒执行

appendfsync no 不主动进行同步，由操作系统进行同步，这是最快，最不安全

文件损坏修复

通过 redis-check-aof -fix

Redis集群

集群方式

1）master/slave

弄三台服务器 192.168.48.133 192.168.48.134 192.168.48.136

俩台从机器 192.168.48.134 192.168.48.136 slave机器配置 slaveof 192.168.48.133 6379

然后我们进去master服务器

master和slave数据是一致

默认情况下slave是只读的，slave上的数据无法同步到master

配置过程 修改48.134和48.136的redis.conf文件，增加slaveof masterip masterport

slaveof 192.168.48.133 6379

实现原理

1. slave第一次或者重连到master上以后，会向master发送一个SYNC的命令
2. master收到SYNC的时候，会做两件事
   a. 执行bgsave（rdb的快照文件）
   b. master会把新收到的修改命令存入到缓冲区
   缺点 没有办法对master进行动态选举【master挂了，就只能读不能写了】

在slave做一个监听，在master中操作redis，我们能在slave中看到相应信息

复制的方式

1. 基于rdb文件的复制（第一次连接或者重连的时候）
2. 无硬盘复制

3. 增量复制

PSYNC master run id. offset

2. 哨兵机制

sentinel

1. 监控master和salve是否正常运行
2. 如果master出现故障，那么会把其中一台salve数据升级为master

配置哨兵

首先我们准备三台redis配置好master和slave模式【配置master/slave具体参考上面内容】

192.168.48.133 master

192.168.48.134 slave

192.168.48.136 slave

在134上复制一份sentinel.conf

修改配置文件中的

sentinel monitor mymaster 192.168.48.133 6379 1

然后启动

然后关闭133的master，我们能够监听出如下内容：

info中显示133由于中断所以哨兵重新选择了134作为master，我们也能在136中可以看到这个信息。

此时我们发现每台slave的配置文件动态修改了

哨兵也是可以集群的。

3）集群【redis3.0之后的功能】

集群原理

Redis Cluster中，Sharding采用slot(槽)的概念，一共分成16384个槽，这有点儿类似前面讲的pre sharding思路。对于每个进入Redis的键值对，根据key进行散列，分配到这16384个slot中的某一个中。使用的hash算法也比较简单，就是CRC16后16384取模。Redis集群中的每个node(节点)负责分摊这16384个slot中的一部分，也就是说，每个slot都对应一个node负责处理。当动态添加或减少node节点时，需要将16384个槽做个再分配，槽中的键值也要迁移。当然，这一过程，在目前实现中，还处于半自动状态，需要人工介入。Redis集群，要保证16384个槽对应的node都正常工作，如果某个node发生故障，那它负责的slots也就失效，整个集群将不能工作。为了增加集群的可访问性，官方推荐的方案是将node配置成主从结构，即一个master主节点，挂n个slave从节点。这时，如果主节点失效，Redis Cluster会根据选举算法从slave节点中选择一个上升为主节点，整个集群继续对外提供服务。这非常类似服务器节点通过Sentinel监控架构成主从结构，只是Redis Cluster本身提供了故障转移容错的能力。

slot（槽）的概念，在redis集群中一共会有16384个槽，

根据key 的CRC16算法，得到的结果再对16384进行取模。 假如有3个节点

node1 0 5460

node2 5461 10922

node3 10923 16383

节点新增

node4 0-1364,5461-6826,10923-12287

删除节点

先将节点的数据移动到其他节点上，然后才能执行删除

市面上提供了集群方案

1. redis shardding 而且jedis客户端就支持shardding操作 SharddingJedis ； 增加和减少节点的问题； pre shardding 3台虚拟机 redis 。但是我部署了9个节点 。每一台部署3个redis增加cpu的利用率 9台虚拟机单独拆分到9台服务器
2. codis基于redis2.8.13分支开发了一个codis-server
3. twemproxy twitter提供的开源解决方案

redis 设置密码 requirepass

redis 缓存的更新 俩个不同的存储，如何保证原子性

1【先删除缓存，在更新数据库】可能出现脏读

2【先更新数据库，更新成功之后，让缓存失效】减少了脏读的可能性，但是还是有概率出现脏读

3【更新数据的时候，只更新缓存，不更新数据库，然后通过异步调度去批量更新数据库】提升性能，但是无法保证强一致性。

关于缓存穿透击穿及失效雪崩的思考和解决方案

1）缓存穿透

概念：缓存穿透是指查询一个一定不存在的数据，由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义。在流量大时，可能DB就挂掉了，要是有人利用不存在的key频繁攻击我们的应用，这就是漏洞。

解决方案【俩种方式】：

1）最常见的是采取布隆过滤器，把所有可能存在的数据哈希到一个足够大的bitmap中，一个不存在的数据会被bitmap拦截掉，避免对底层存储系统的查询压力。

2）如果查询为空，把这个空结果缓存，过期时间设置不超过5分钟。

2）缓存击穿

概念：对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题：缓存被“击穿”的问题，这个和缓存雪崩的区别在于这里针对某一key缓存，前者则是很多key。

缓存在某个时间点过期的时候，恰好在这个时间点对这个Key有大量的并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。

解决方案：

1）使用互斥锁

2）提前使用互斥锁

3）不过期，异步构建缓存，不会阻塞线程池

4）资源隔离组件hystrix

3）缓存失效

概念：缓存雪崩是指在我们设置缓存时采用了相同的过期时间，导致缓存在某一时刻同时失效，请求全部转发到DB，DB瞬时压力过重雪崩。

解决方案：

缓存失效时的雪崩效应对底层系统的冲击非常可怕。大多数系统设计者考虑用加锁或者队列的方式保证缓存的单线 程（进程）写，从而避免失效时大量的并发请求落到底层存储系统上。这里分享一个简单方案就时讲缓存失效时间分散开，比如我们可以在原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。