前言
这次的内容是我自己为了总结Redis知识而扩充的，但是Redis的强大，以及适用范围之广可不是单单一篇博文就能总结清的。所以这次准备继续总结，因为第一个问题，Redis的批量操作，是我在面试过程中被真实问到的，当时没答上来，也是因为确实没了解过Redis的批量操作。
当时的问题，我还记得比较清晰：Redis执行批量操作的功能是什么？使用场景就是搞促销活动时，会做预缓存，会往缓存里放大批数据，如果直接放的话那么会很慢，怎么能提高效率呢？
Redis的批量操作-管道（pipeline）
首先Redis的管道（pipeline）并不是Redis服务端提供的功能，而是Redis客户端为了减少网络交互而提供的一种功能。
正常的一次Redis网络交互如下：
pipeline主要就是将多个请求合并，进行一次提交给Redis服务器，Redis服务器将所有请求处理完成之后，再一次性返回给客户端。
下面我们分析一下 pipeline 的原理
pipeline的一个交互过程是这样的：
- 客户端进程调用 write 命令将消息写入到操作系统内核为套接字分配的 发送缓冲区send buffer 。
- 客户端操作系统通过网络路由，将 send buffer 中的数据发送给服务器操作系统为套接字分配的接 收缓冲区 receive buffer 。
- 服务端进程调用 read 命令从 receive buffer 中取出数据进行处理，然后调用 write命令将相应信息写入到服务端的 send buffer 中。
- 服务端操作系统通过网络路由，将 send buffer 中的数据发送给客户端操作系统的receive buffer 。
- 客户端进程调用read命令将数据从 receive buffer 中取出进行业务处理。
在使用 pipeline 时需要注意：
- pipeline执行的操作，和mget，mset，hmget这样的操作不同，pipeline的操作是不具备原子性的。
- 还有在集群模式下因为数据是被分散在不同的slot里面的，因此在进行批量操作的时候，不能保证操作的数据都在同一台服务器的slot上，所以集群模式下是禁止执行像mget、mset、pipeline等批量操作的，如果非要使用批量操作，需要自己维护key与slot的关系。
- pipeline也不能保证批量操作中有命令执行失败了而中断，也不能让下一个指令依赖上一个指令，如果非要这样的复杂逻辑，建议使用lua脚本来完成操作。
Redis实现消息队列和延时队列
消息队列
Redis的实现消息队列可以用list来实现，通过lpush与rpop或者rpush与lpop结合来实现消息队列。
但是若是list为空后，无论是lpop还是rpop都会持续的获取list中的数据，若list一直为空，持续的拉取数据，一是会增加客户端的cpu利用率，二是也增高了Redis的QPS，解决方案是使用 blpop或 brpop 来代替lpop或rpop。
其实blpop和brpop的作用是bloking pop，就是阻塞拉取数据，当消息队列中为空时就会停止拉取，有数据后立即恢复拉取。
但是当没有数据的时候， 阻塞拉取 ，就会一直阻塞在那里，时间久了就成了空闲连接，那么Redis服务器一般会将时间闲置过久的连接直接断掉，以减少连接资源。所以还要检测 阻塞拉取 抛出的异常然后进行重试。
另外一点，就是Redis实现的消息队列，没有ACK机制，所以想要实现消息的可靠性，还要自己实现当消息处理失败后，能继续抛回队列。
延时队列
用Redis实现延时队列，其实就是使用zset来实现，将消息序列化成一个字符串（可以是json格式），作为为 value ，消息的到期处理时间做为 score ，然后用多线程去查询zset来获取到期消息进行处理。
多线程轮询处理，保证了可用性，但是要做幂等或锁处理，保证不要重复处理消息。
主要的实现代码如下。
/**
* 放入延时队列
* @param queueMsg
*/
private void delay(QueueMsg queueMsg){
String msg = JSON.toJSONString(queueMsg);
jedis.zadd(queueKey,System.currentTimeMillis()+5000,msg);
}
/**
* 处理队列中从消息
*/
private void lpop(){
while (!Thread.interrupted()){
// 从队列中取出，权重为0到当前时间的数据，并且数量只取一个
Set<String> strings = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
// 如果消息为空，就歇会儿再取。
if(strings.isEmpty()){
try {
//休息一会儿
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
break;
}
continue;
}
String next = strings.iterator().next();
// 如果抢到了消息
if(jedis.zrem(queueKey,next)>0){
// 反序列化后获取到消息
QueueMsg queueMsg = JSON.parseObject(next, QueueMsg.class);
// 进行消息处理
handleMsg(queueMsg);
}
}
}
订阅模式
Redis的主题订阅模式，其实并不想过多总结，因为由于它本身的一些缺点，导致它的应用场景比较窄。
前面总结的用Redis的list实现的消息队列，虽然可以使用，但是并不支持消息多播的场景，即一个生产者，将消息放入到多个队列中，然后多个消费者进行消费。
这种消息多播的场景常用来做分布式系统中的解耦。用哦 publish 进行生产者发送消息，消费者使用 subscribe 进行获取消息。
例如：我向jimoerChannel发送了一条消息 b-tree
127.0.0.1:6379> publish jimoerChannel b-tree
(integer) 1
订阅这个渠道的消费者立马收到了一条b-tree的消息。
127.0.0.1:6379> subscribe jimoerChannel
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "jimoerChannel"
3) (integer) 1
1) "message"
2) "jimoerChannel"
3) "b-tree"
我前面也说到了，Redis的pub/sub订阅模式，其实最大的缺点就是，消息不能持久化，这样就导致，若是消费者挂了或是没有消费者，那么消息就会被直接丢弃。因为这个原因，所以导致他的使用场景比较少。
IO模型
Redis的过期策略
Redis的过期策略是适用于所有数据结构的。数据一到过期时间就自动删除，Redis会将设置了过期时间的key 放置在一个字典表里。
定期删除
Redis会定期遍历字典表里面数据来删除过期的Key。
Redis默认的定期删除策略是每秒进行10次过期扫描，即每100ms扫描一次。并不是扫描全部设置了过期时间的key，而是随机扫描20个key，删除掉已经过期的key，如果过期的比率超过25%，那么就继续进行扫描。
惰性删除
因为定期删除是随机抽取一些key来进行过期删除，所以如果key并没有被定期扫描到，那么过期的key就不会被删除。所以Redis还提供了惰性删除的策略， 就是当去查询某些key的时候，若是key已经过期了，那么就会删除key，然后返回null。
另外一点当在集群条件下，主从同步情况中，主节点中的key过期后，会在aof中生成一条删除指令，然后同步到从节点，这样的从节点在接收到aof的删除指令后，删除掉从节点的key，因为主从同步的时候是异步的所以，短暂的会出现主节点已经没有数据了，但是从节点还存在。
但是若是定期删除也没有扫描到key，而且好长时间也没去去使用key，那么这部分过期的key就会一直占用的内存。
所以Redis又提供了内存淘汰机制。
内存淘汰机制
当Redis的内存出现不足时，就会持续的和磁盘进行交互，这样就会导致Redis卡顿，效率降低等情况。这在线上是不允许发生的，所以Redis提供了配置参数 maxmemory 来限制内存超出期望大小。
当内存使用情况超过maxmemory的值时，Redis提供了以下几种策略，来让使用者通过配置决定该如何腾出内存空间来继续提供服务。
- **noeviction **不会继续提供写请求（del请求可以），读请求可以，写请求会报错，这样保证的数据不会丢失，但是业务不可用，这是默认的策略。
- **volatile-lru **会将设置了过期时间的key中，淘汰掉最近最少使用的key。没有设置过期时间的key不会被淘汰，保证了需要持久化的数据不丢。
- **volatile-ttl **尝试将设置了过期时间的key中，剩余生命周期越短，越容易被淘汰。
- **volatile-random **尝试将从设置了过期时间的key中，随机选择一些key进行淘汰。
- **allkeys-lru **从所有key中，淘汰掉最近最少使用的key。
- **allkeys-random **从所有key中，随机淘汰一部分key。
那么具体设置成哪种淘汰策略呢？
这就是要看在使用Redis时的具体场景了，如果只是用Redis做缓存的话，那么可以配置allkeys-lru或allkey-random，客户端在写缓存的时候并不用携带着过期时间。若是还想要用持久化的功能，那么就应该使用volatile-开头的策略，这样可以保证每月设置过期时间的key不会被淘汰。
内存淘汰策略的配置如下：
# 最大使用内存
maxmemory 5m
# 内存淘汰策略 The default is:noeviction
maxmemory-policy allkeys-lru
LRU算法
LRU算法的实现，其实可以靠一个链表。链表按照使用情况来进行排序，当空间不足时，会剔除掉尾部的数据。当某个元素被访问时它会被移动到链表头。
在真实的面试中，若是让写出LRU算法，我认为可以使用Java中的LikedHashMap来实现，因为LikedHashMap已经实现了基本的LRU功能，我只需要封装一下就改造成了自己的了。
/**
* @author Jimoer
* @description
*/
public class MyLRUCache<K,V> {
// lru容量
private int lruCapacity;
// 数据容器（内存）
private Map<K,V> dataMap;
public MyLRUCache(int capacity){
this.lruCapacity = capacity;
// 设置LinkedHashMap的初始容量为LRU的最大容量，
// 扩容因子为默认的0.75，第三个参数是否将数据按照访问顺序排序。
dataMap = new LinkedHashMap<K, V>(capacity, 0.75f, true){
@Override
protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
// 当数据量大于lruCapacity时，移除掉最老使用的数据。
return super.size()>lruCapacity;
}
};
}
public V get(K k){
return dataMap.get(k);
}
public void put(K key, V value){
dataMap.put(key,value);
}
public int getLruCapacity() {
return lruCapacity;
}
public Map<K, V> getDataMap() {
return dataMap;
}
}
测试代码：
@Test
public void lruTest(){
// 内存容量为3，即存储3条数据后，再放入数据，就会将最老使用的数据删除
MyLRUCache myLRUCache = new MyLRUCache(3);
myLRUCache.put("1k","张三");
myLRUCache.put("2k","李四");
myLRUCache.put("3k","王五");
// 容量已满
System.out.println("myLRUCache:"+JSON.toJSONString(myLRUCache.getDataMap()));
// 继续放入数据，该删除第一条数据为第四条数据腾出空间了
myLRUCache.put("4k","赵六");
// 打印出结果
System.out.println("myLRUCache:"+JSON.toJSONString(myLRUCache.getDataMap()));
}
运行结果：
myLRUCache:{"1k":"张三","2k":"李四","3k":"王五"}
myLRUCache:{"2k":"李四","3k":"王五","4k":"赵六"}
总结
好了，Redis的相关知识，就总结到这里了，算上前面两篇博文（ Redis基础数据结构总结 、 你说一下Redis为什么快吧，怎么实现高可用，还有持久化怎么做的 ），这是Redis的第三篇了，这一篇博文也是新年的第一篇，元旦假期在家花了两天时间，自己学习自己总结。元旦假期结束后，我要继续面试了，后面我会继续将我面试中遇到的各种问题，总结出来，一是增加自己的知识面，二也将知识进行的传播。
希望可以对大家学习Redis有帮忙，喜欢的小伙伴可以帮忙转发+关注，感谢大家！
