redis 基础知识

Java缓存机制

1、类似于static HashMap

2、保存对象的有效性以及周期无法控制，这样很容易就导致内存急剧上升

3、常用的有Oscache,Ehcache,Jcache,Jbosscache

4、ehcache 主要是对数据库访问的缓存、oscache 主要是对页面的缓存

Redis介绍

1、非关系型的数据库

2、Redis支持数据的持久化，可以将数据存放在硬盘上。

3、提供string、list、set、zset、hash等数据结构的存储

4、支持主从复制（数据备份、读写分离、集群、高可用、宕机容错）

5、应用场景 解决数据库的访问压力。

6、所有操作都是原子性的，支持事务操作。

7、Redis为单进程单线程模式

String常用命令

[code]SET key value	此命令设置指定键的值。
GET key	获取指定键的值。
GETRANGE key start end	获取存储在键上的字符串的子字符串。
GETSET key value	设置键的字符串值并返回其旧值。
GETBIT key offset	返回在键处存储的字符串值中偏移处的位值。
MGET key1 [key2..]	获取所有给定键的值
SETBIT key offset value	存储在键上的字符串值中设置或清除偏移处的位
SETEX key seconds value	使用键和到期时间来设置值
SETNX key value	设置键的值，仅当键不存在时
SETRANGE key offset value	在指定偏移处开始的键处覆盖字符串的一部分
STRLEN key	获取存储在键中的值的长度
MSET key value [key value …]	为多个键分别设置它们的值
MSETNX key value [key value …]	为多个键分别设置它们的值，仅当键不存在时
PSETEX key milliseconds value	设置键的值和到期时间(以毫秒为单位)
INCR key	将键的整数值增加1
INCRBY key increment	将键的整数值按给定的数值增加
INCRBYFLOAT key increment	将键的浮点值按给定的数值增加
DECR key	将键的整数值减1
DECRBY key decrement	按给定数值减少键的整数值
APPEND key value	将指定值附加到键

list常用命令

[code]BLPOP key1 [key2 ] timeout 移出并获取列表的第一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。
BRPOP key1 [key2 ] timeout 移出并获取列表的最后一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。
BRPOPLPUSH source destination timeout 从列表中弹出一个值，将弹出的元素插入到另外一个列表中并返回它； 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。
LINDEX key index 通过索引获取列表中的元素
LINSERT key BEFORE|AFTER pivot value 在列表的元素前或者后插入元素
LLEN key 获取列表长度
LPOP key 移出并获取列表的第一个元素
LPUSH key value1 [value2] 将一个或多个值插入到列表头部
LPUSHX key value 将一个值插入到已存在的列表头部
LRANGE key start stop 获取列表指定范围内的元素
LREM key count value 移除列表元素
LSET key index value 通过索引设置列表元素的值
LTRIM key start stop 对一个列表进行修剪(trim)，就是说，让列表只保留指定区间内的元素，不在指定区间之内的元素都将被删除。
RPOP key 移除并获取列表最后一个元素
RPOPLPUSH source destination 移除列表的最后一个元素，并将该元素添加到另一个列表并返回
RPUSH key value1 [value2] 在列表中添加一个或多个值
RPUSHX key value 为已存在的列表添加值

 set常用命令

[code]SADD key member1 [member2] 向集合添加一个或多个成员
SCARD key 获取集合的成员数
SDIFF key1 [key2] 返回给定所有集合的差集
SDIFFSTORE destination key1 [key2] 返回给定所有集合的差集并存储在 destination 中
SINTER key1 [key2] 返回给定所有集合的交集
SINTERSTORE destination key1 [key2] 返回给定所有集合的交集并存储在 destination 中
SISMEMBER key member 判断 member 元素是否是集合 key 的成员
SMEMBERS key 返回集合中的所有成员
SMOVE source destination member 将 member 元素从 source 集合移动到 destination 集合
SPOP key 移除并返回集合中的一个随机元素
SRANDMEMBER key [count] 返回集合中一个或多个随机数
SREM key member1 [member2] 移除集合中一个或多个成员
SUNION key1 [key2] 返回所有给定集合的并集
SUNIONSTORE destination key1 [key2] 所有给定集合的并集存储在 destination 集合中
SSCAN key cursor [MATCH pattern] [COUNT count] 迭代集合中的元素

sorted set

Hash

Redis持久化

1、AOF 与 RDB 两种模式

RDB 是在某个时间 点将数据写入一个临时文件，持久化结束后，用这个临时文件替换上次持久化的文件，达到数据恢复。
优点：使用单独子进程来进行持久化，主进程不会进行任何 IO 操作，保证了 redis 的高性能
缺点：RDB 是间隔一段时间进行持久化，如果持久化之间 redis 发生故障，会发生数据丢失。所以这种方式更适合数据要求不严谨的时候
这里说的这个执行数据写入到临时文件的时间点是可以通过配置来自己确定的，通过配置redis 在 n 秒内如果超过 m 个 key 被修改这执行一次 RDB 操作。这个操作就类似于在这个时间点来保存一次 Redis 的所有数据，一次快照数据。所有这个持久化方法也通常叫做 snapshots。

Append-only file，将“操作 + 数据”以格式化指令的方式追加到操作日志文件的尾部，在 append 操作返回后(已经写入到文件或者即将写入)，才进行实际的数据变更，“日志文件”保存了历史所有的操作过程；当 server 需要数据恢复时，可以直接 replay 此日志文件，即可还原所有的操作过程。AOF 相对可靠，它和 mysql 中 bin.log、apache.log、zookeeper 中 txn-log 简直异曲同工。AOF 文件内容是字符串，非常容易阅读和解析。
优点：可以保持更高的数据完整性，如果设置追加 file 的时间是 1s，如果 redis 发生故障，最多会丢失 1s 的数据；且如果日志写入不完整支持 redis-check-aof 来进行日志修复；AOF 文件没被 rewrite 之前（文件过大时会对命令进行合并重写），可以删除其中的某些命令（比如误操作的 flushall）。
缺点：AOF 文件比 RDB 文件大，且恢复速度慢。

我们可以简单的认为 AOF 就是日志文件，此文件只会记录“变更操作”(例如：set/del 等)，如果 server 中持续的大量变更操作，将会导致 AOF 文件非常的庞大，意味着 server 失效后，数据恢复的过程将会很长；事实上，一条数据经过多次变更，将会产生多条 AOF 记录，其实只要保存当前的状态，历史的操作记录是可以抛弃的；因为 AOF 持久化模式还伴生了“AOF rewrite”。
AOF 的特性决定了它相对比较安全，如果你期望数据更少的丢失，那么可以采用 AOF 模式。如果 AOF 文件正在被写入时突然 server 失效，有可能导致文件的最后一次记录是不完整，你可以通过手工或者程序的方式去检测并修正不完整的记录，以便通过 aof 文件恢复能够正常；同时需要提醒，如果你的 redis 持久化手段中有 aof，那么在 server 故障失效后再次启动前，需要检测 aof 文件的完整性。

高可用

Redis 的 Sentinel 系统用于管理多个 Redis 服务器（instance）， 该系统执行以下三个任务：

• 监控（Monitoring）： Sentinel 会不断地检查你的主服务器和从服务器是否运作正常。
• 提醒（Notification）： 当被监控的某个 Redis 服务器出现问题时， Sentinel 可以通过 API 向管理员或者其他应用程序发送通知。
• 自动故障迁移（Automatic failover）： 当一个主服务器不能正常工作时， Sentinel 会开始一次自动故障迁移操作， 它会将失效主服务器的其中一个从服务器升级为新的主服务器， 并让失效主服务器的其他从服务器改为复制新的主服务器； 当客户端试图连接失效的主服务器时， 集群也会向客户端返回新主服务器的地址， 使得集群可以使用新主服务器代替失效服务器。

Redis Sentinel 是一个分布式系统， 你可以在一个架构中运行多个 Sentinel 进程（progress）， 这些进程使用流言协议（gossip protocols)来接收关于主服务器是否下线的信息， 并使用投票协议（agreement protocols）来决定是否执行自动故障迁移， 以及选择哪个从服务器作为新的主服务器。

虽然 Redis Sentinel 释出为一个单独的可执行文件 redis-sentinel ， 但实际上它只是一个运行在特殊模式下的 Redis 服务器， 你可以在启动一个普通 Redis 服务器时通过给定 –sentinel 选项来启动 Redis Sentinel 。

Redis分布式锁 

1、原因：为了保证一个方法在高并发情况下的同一时间只能被同一个线程执行

2、使用条件：系统是一个分布式系统、共享资源、同步访问

3、应用的场景：管理后台的部署架构（多台tomcat服务器+redis【多台tomcat服务器访问一台redis】+mysql【多台tomcat服务器访问一台服务器上的mysql】）就满足使用分布式锁的条件。多台服务器要访问redis全局缓存的资源，如果不使用分布式锁就会出现问题。

5、实现方式：

• setNX（key:value）：如果value存在返回失败0，不存在返回成功1（value是过期时间）
• get：获取当前的value值
• getSET（key:set）: 设置新的值时返回老的值
• 解决死锁：使用超时

   流程：任务 -》尝试setNX设置-》失败的情况下，get 看时间是否过期-》过期使用getSET，如果获取到的仍然是过期时间，带代表锁获取成功，执行自己的任务。获取到的不是过期时间继续等待

[code]package test.miaosha;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

public class RedisLock {

private static Logger logger = LoggerFactory.getLogger(RedisLock.class);

private RedisTemplate<String,Object> redisTemplate;

private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;

/**
* Lock key path.
*/
private String lockKey;

/**
* 锁超时时间，防止线程在入锁以后，无限的执行等待
*/
private int expireMsecs = 60 * 1000;

/**
* 锁等待时间，防止线程饥饿
*/
private int timeoutMsecs = 10 * 1000;

private volatile boolean locked = false;

/**
* Detailed constructor with default acquire timeout 10000 msecs and lock
* expiration of 60000 msecs.
*
* @param lockKey
*            lock key (ex. account:1, ...)
*/
public RedisLock(RedisTemplate<String,Object> redisTemplate, String lockKey) {
this.redisTemplate = redisTemplate;
this.lockKey = lockKey + "_lock";
}

/**
* Detailed constructor with default lock expiration of 60000 msecs.
*
*/
public RedisLock(RedisTemplate<String,Object> redisTemplate, String lockKey, int timeoutMsecs) {
this(redisTemplate, lockKey);
this.timeoutMsecs = timeoutMsecs;
}

/**
* Detailed constructor.
*
*/
public RedisLock(RedisTemplate<String,Object> redisTemplate, String lockKey, int timeoutMsecs, int expireMsecs) {
this(redisTemplate, lockKey, timeoutMsecs);
this.expireMsecs = expireMsecs;
}

/**
* @return lock key
*/
public String getLockKey() {
return lockKey;
}

public String get(final String key) {
Object obj = null;
try {
obj = redisTemplate.execute(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
StringRedisSerializer serializer = new StringRedisSerializer();
byte[] data = connection.get(serializer.serialize(key));
connection.close();
if (data == null) {
return null;
}
return serializer.deserialize(data);
}
});
} catch (Exception e) {
logger.error("get redis error, key : {}", key);
}
return obj != null ? obj.toString() : null;
}

public String set(final String key,final String value) {
Object obj = null;
try {
obj = redisTemplate.execute(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
StringRedisSerializer serializer = new StringRedisSerializer();
connection.set(serializer.serialize(key), serializer.serialize(value));
return serializer;
}
});
} catch (Exception e) {
logger.error("get redis error, key : {}", key);
}
return obj != null ? obj.toString() : null;
}

public boolean setNX(final String key, final String value) {
Object obj = null;
try {
obj = redisTemplate.execute(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
StringRedisSerializer serializer = new StringRedisSerializer();
Boolean success = connection.setNX(serializer.serialize(key), serializer.serialize(value));
connection.close();
return success;
}
});
} catch (Exception e) {
logger.error("setNX redis error, key : {}", key);
}
return obj != null ? (Boolean) obj : false;
}

private String getSet(final String key, final String value) {
Object obj = null;
try {
obj = redisTemplate.execute(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
StringRedisSerializer serializer = new StringRedisSerializer();
byte[] ret = connection.getSet(serializer.serialize(key), serializer.serialize(value));
connection.close();
return serializer.deserialize(ret);
}
});
} catch (Exception e) {
logger.error("setNX redis error, key : {}", key);
}
return obj != null ? (String) obj : null;
}

/**
* 获得 lock. 实现思路: 主要是使用了redis 的setnx命令,缓存了锁. reids缓存的key是锁的key,所有的共享,
* value是锁的到期时间(注意:这里把过期时间放在value了,没有时间上设置其超时时间) 执行过程:
* 1.通过setnx尝试设置某个key的值,成功(当前没有这个锁)则返回,成功获得锁
* 2.锁已经存在则获取锁的到期时间,和当前时间比较,超时的话,则设置新的值
*
* @return true if lock is acquired, false acquire timeouted
* @throws InterruptedException
*             in case of thread interruption
*/
public synchronized boolean lock() throws InterruptedException {
int timeout = timeoutMsecs;
while (timeout >= 0) {
long expires = System.currentTimeMillis() + expireMsecs + 1;
String expiresStr = String.valueOf(expires); // 锁到期时间
if (this.setNX(lockKey, expiresStr)) {
// lock acquired
locked = true;
return true;
}

String currentValueStr = this.get(lockKey); // redis里的时间
if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
// 判断是否为空，不为空的情况下，如果被其他线程设置了值，则第二个条件判断是过不去的
// lock is expired

String oldValueStr = this.getSet(lockKey, expiresStr);
// 获取上一个锁到期时间，并设置现在的锁到期时间，
// 只有一个线程才能获取上一个线上的设置时间，因为jedis.getSet是同步的
if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
// 防止误删（覆盖，因为key是相同的）了他人的锁——这里达不到效果，这里值会被覆盖，但是因为什么相差了很少的时间，所以可以接受

// [分布式的情况下]:如过这个时候，多个线程恰好都到了这里，但是只有一个线程的设置值和当前值相同，他才有权利获取锁
// lock acquired
locked = true;
return true;
}
}
timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;

/*
* 延迟100 毫秒, 这里使用随机时间可能会好一点,可以防止饥饿进程的出现,即,当同时到达多个进程,
* 只会有一个进程获得锁,其他的都用同样的频率进行尝试,后面有来了一些进行,也以同样的频率申请锁,这将可能导致前面来的锁得不到满足.
* 使用随机的等待时间可以一定程度上保证公平性
*/
Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);

}
return false;
}

/**
* Acqurired lock release.
*/
public synchronized void unlock() {
if (locked) {
redisTemplate.delete(lockKey);
locked = false;
}
}

}