之前我们介绍了一些列关于Redis的数据结构、持久化、过期&淘汰策略、集群化等知识点，感兴趣的小伙伴可以在文章的末尾查看往期内容。今天将为大家带来Redis的应用。由于本篇文章较长，所以将拆分为两章来讲解。
除了最基本的KV缓存外，Redis还能用到以下方面。
分布式锁
在分布式应用中，经常会遇到并发问题需要处理，这个时候往往就需要用到分布式锁，而Redis很好的提供了分布式锁的处理。常用的Redis分布式锁有以下两种方式。
setnx
-
setnx（set if not exists）：通过setnx设置key，达到获取锁的目的。处理完逻辑通过del删除key释放锁。
-
死锁处理：为了避免释放锁失败，导致程序进入死锁，通常采用 给key加上过期时间的方式来处理。
Redis2.8 开始加入了set扩展参数，支持setnx 和 expire可以同时执行，是setnx 和 expire在同一个事务中，避免了expire设置过期时间不成功的问题。
- 任务超时：设置超时时间虽然能避免死锁，但是会存在因超时而使锁提前释放的可能。所以该方案不适合需要较长时间的任务。为了避免任务超时，可以采取 给key设置随机value的方式来解决该问题。释放锁时，先匹配value是否一致，再删除key。
Redlock
除了setnx外，Redis还提供了Redlock提供Redis分布式锁。
要求：Redlock需要多个Redis实例支持，这些Redis实例要求相互独立，没有主从关系。
实现：加锁时，客户端会向超过半数的节点发送setnx指令，只要过半数节点set成功，则加锁成功。释放锁时，需要向所有节点发送del指令。
使用场景：Redlock适用于对高可用要求高的场景，即使一台Redis挂了，也不会影响分布式锁的实现。同时Redlock需要更多的Redis实例，相较于setnx性能有所下降，同时运维代价更高。
消息队列
Redis中实现消息队列常用的list、zset 及Stream几种数据结构可以实现。
list
Redis可以通过list实现异步消息队列。
-
实现：用rpush和lpush实现入队操作，lpop和rpop实现出队操作，rpush与lpop，lpush与rpop配合使用；
-
阻塞读：lpop/rpop存在空队列循环取值的问题，造成不必要的资源浪费。通过使用blpop/brpop可以实现队列的阻塞读。当队列中没有数据时，读操作会进入休眠等待状态，当有数据时，则立即唤醒读操作。
zset
Redis通过zset可以实现延时队列。
-
实现：将消息内容序列化成字符串，作为zset的value，消息处理的到期时间作为score。然后轮询zet获取到期的任务进行处理。
-
zrem：Redis的zrem(key, value)指令，可以从zset中移除指定value，同时能够避免多线程、多进程之间的资源争抢。
Stream
Redis自5.0起开始支持Stream数据结构，Stream是一个能够支持多播可持久化化的消息队列。
- 实现：
Stream结构.png
- Stream为一个消息链表，每个节点包含有一个唯一的消息ID和消息内容两部分内容，消息是持久化的；
- 每个Stream都可以设置多个消费组组，每个消费组都有一个游标(last_delivered_id)，表示当前消费组已经消费到了哪条消息，各个消费组间互不影响；
- 每个消费组都可以有多个消费者，消费者之间是相互竞争，即同一条消息在同一个消费组内不会被重复消费；
- PEL（Pending Entries List） —— 每个消费组内都有一个pending_ids，记录当前哪些消息已被客户端读取，当收到客户端返回的ack时，消息ID从pending_ids中排除。以此结构来确保客户端至少消费了一次消息，保证消息不会被丢失。
- 指令：Steram有以下常用的指令：
- xadd: 向Stream中添加消息；
// * 表示服务器自动生成ID，后面分别是 key value
// maxlen 表示设置该stream的最大长度，当消息过长，会丢弃老的消息
127.0.0.1:6379> xadd maxlen 100 javamd * test null
1582983245000-0 // 返回生成的消息ID，1 表示当前时间生成的第一条消息
- xdel: 从Stream中逻辑移除消息（设置标记位，并不影响Stream的真实长度）；
127.0.0.1:6379> xdel javamd 1582983245000-1
(integer) 1
- xrange：获取Stream中未标记的消息列表；
127.0.0.1:6379> xrange javamd 1582983245000-0 + // 1582983245000-0为最小消息ID的列表
1) 1) 1582983245000-2
2) 1) “test”
2）null
1) 1) 1582983245000-0
2) 1) “test”
2）null
127.0.0.1:6379> xrange javamd - 1582983245000-8 // 1582983245000-8为最大消息ID的列表
1) 1) 1582983245000-0
2) 1) “test”
2）null
2) 1) 1582983245000-2
2) 1) “test”
2）null
- xlen: 获取Stream中所有消息的长度；
127.0.0.1:6379> xlen javamd
(integer) 8 // 不包含已标记的消息
- del：物理删除整个Steram；
127.0.0.1:6379> del javamd
(integer) 1 // 删除整个Stream
- xread：在不定义消费组的情况下进行Stream的独立消费；
127.0.0.1:6379> xread count 2 streams javamd 0-0 // 从Stream头部读取两条消息
1) 1) “javamd”
2） 1) 1) 1582983245000-0
2) 1) “test”
2）null
2) 1) 1582983245000-2
2) 1) “test”
2）null
使用xread进行顺序消费时，需要记录当前消费消息的ID。下次调用时，需要将当前消费的消息ID作为参数回传。
xread 可以添加block参数，当参数值大于0时，表示阻塞多少ms，如果相应时间内没有消息到来，返回nil。当参数等于0时，表示永远阻塞，直到新的消息添加到Stream中。
- xgroup create: 创建消费组;
127.0.0.1:6379> xgroup create javamd group_javamd1 0-0 // 表示从头部开始消费
OK
127.0.0.1:6379> xgroup create javamd group_javamd2 $ // $ 表示从尾部开始消费，只接受新消息
OK
- xreadgroup：通过消费组进行消费。
// > 表示从当前消费组的 last_delivered_id 后面开始读
127.0.0.1:6379> xreadgroup GROUP javamd1 c1 count 1 streams javamd >
1) 1) “javamd”
2） 1) 1) 1582983245000-0
2) 1) “test”
2）null
未完待续……
网友评论