Redis简单概述：今天主要简单聊聊Redis在工作中的一些应用，有说的不对的地方勿拍砖啊。说到Redis，可能有不少朋友会说它就是一个缓存数据库，没错它确实主要是干缓存这件事，在我之前仅用过它的String或者再多一点Hash这两结构的时候，我也一度觉这么认为。再后来因为工作需要，接触到了它其他的一些结构，List、Set等等以及底层一些实现，回过头来突然发现它完全就是迎合互联网市场的，这还只是应用层，在底层方面，比如IO模型、服务模型、数据结构，算法设计等等。保守点说，现在还有哪个互联网公司缓存这块没用Redis的？更甚至过分的说，有些业务模块直接用它来当数据库存储都有不少。这么牛逼的产品，到底是哪位大神写的？Redis之父（Salvatore Sanfilippo），就是这个意大利神人，太强了，拜服啊。   分布式锁：先简单讨论下为啥要有锁？服务端天生就是一个多线程环境（要不叫服务），这个特点跟CS客户端有本质区别，CS客户端如果你不主动开启work现场，它自始至终只有一个线程在跑，那就是UI线程（主线程）。说的有点跑题了啊，那么多线程环境有个特点，天生会导致资源不同步。在现实业务中，比如减库存操作，有3个客户下单同时读取了库存数据都是100，每人购买一件并且发生减库存操作，最终库存还是99，那么这就出现了BUG。这还只是单机环境，如果是集群环境呢？肯定只会更复杂，我们这里不讨论单机环境，我相信现在没有人会做单机部署吧，即便就算是单机（最少要有高可用集群），我相信你也不用考虑并发问题，因为你压根就没有这样的需求。要解决这个问题并非只有分布式锁能处理，比如数据库层面也可以处理，乐观锁&悲观锁都是可以的，只是这样的话，我相信你业务系统的性能和存储系统抗不了几下。好了言归正传，多线程、多进程、多服务器，导致库存数据错误的根本问题在于并行，这是问题的本质，那么我们只要想办法把他们串行起来不就问题解决了么？如果串行，问题又来了，性能差，既然都上分布式了，肯定有高并发需求，那没办法，生产环境宁愿牺牲性能，也好过有致命BUG吧，性能问题可以再优化嘛。那么回到原来的问题，在分布式环境下，要想实现串行，只能借助三方共享服务实现了，在现阶段还是有不少产品服务供我们选择，比如Zookeeper、MQ、还有Redis等等。这里需要注意MQ是异步的方案，结合实际业务吧，可以做相应补偿机制，如京东下单，即便库存不足了，我还可以临近仓库调货。下面我们简单讨论下，用Redis来实现分布式锁，如果我们自己通过Setnx命令或者其他结构来实现一把生产环境能用的分布式锁，还是有点麻烦的，它的麻烦点主要有两个，Redis命令的原子性和锁赎命，主要是这个赎命的问题。有这么一个逻辑，客户端1设置的这把锁需要有超时时间，如果没有这个时间，客户端1挂了，这个KEY对应的锁也就一直挂在那，除非Redis内存不够用淘汰或者手动删除，如果设置时间，这个时间多长合适？10s、100s？这个固定不变的时间都不合适，比如客户端1因为网络资源处理了101s，这把锁被Redis服务端删除了，在并发情况下，其他线程或者进程获得了锁，数据也就可能会有问题了，所以我们看下Redisson开源框架是怎么处理的。看代码：  

"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);",

  上面这段代码我是直接从Redisson源码里面copy出来的。Redisson框架是Java写的，目前好像只有JavaSDK，当然.NET客户端StackExchange.Redis LockTake好像也可以实现，具体没去看它的源码。这是一段加锁代码，这里大概解释一下，这是一段LUA脚本，Redis可以直接执行LUA脚本并且是原子操作，每个客户端都提供有这个API。以上代码的逻辑是， 1.先执行exists命令判断key是否存在（也就是是否存在锁资源），如果为true，则写入hash，key是之前传过来的资源标识符，field为线程id，这里设置了一个value为1，并且设置过期时间。 2.如果hash里面存在上面资源对应的这个数据（也就是自己的这把锁），则执行incrby操作+1，为啥+1？因为这是一把可重入锁，并且重新设置过期时间。 3.如果以上都不是，也就是这个资源有锁并且不是当前线程的资源（也就是有线程已经加锁，正在处理），则直接返回剩余过期时间。 4.以上都是原子操作。这么看来枷锁逻辑是没问题，下面我们看下正常情况下，解锁逻辑，也就是客户端处理完业务，自己解锁。看代码：  

 1 f (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
 2 "return nil;" +
 3 "end; " +
 4 "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
 5 "if (counter > 0) then " +
 6 "redis.call('pexpire', KEYS[1], ARGV[2]); " +
 7 "return 0; " +
 8 "else " +
 9 "redis.call('del', KEYS[1]); " +
10 "redis.call('publish', KEYS[2], ARGV[1]); " +
11 "return 1; " +
12 "end; " +
13 "return nil;",

  同样是一段LUA脚本，看样子要实现Redis的高级功能，LUA脚本是必须的，下面简单说明下，这段正常释放锁的脚本逻辑。 1.这把锁不是当前线程的，无权删除，直接返回。 2.因为这是一把可重入锁，所以先做-1操作，再判断是否为0，如果不为0，重新设置过期时间，如果为0，删除这把锁，并且publish，因为在之前Trylock有订阅这个消息。到这里正常情况下，这个分布式锁逻辑是没有问题，如果非正常情况列？比如，线程业务操作时间超过了过期时间呢？我们继续看下代码：

 1 private void renewExpiration() {
 2 ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
 3 if (ee == null) {
 4 return;
 5 }
 6  
 7 Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
 8 @Override
 9 public void run(Timeout timeout) throws Exception {
10 ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
11 if (ent == null) {
12 return;
13 }
14  
15 Long threadId = ent.getFirstThreadId();
16 if (threadId == null) {
17 return;
18 }
19  
20 RFuture<Boolean> future = renewExpirationAsync(threadId);
21 future.onComplete((res, e) -> {
22 if (e != null) {
23 log.error("Can't update lock " + getRawName() + " expiration", e);
24 EXPIRATION_RENEWAL_MAP.remove(getEntryName());
25 return;
26 }
27  
28 if (res) {
29 // reschedule itself
30 renewExpiration();
31 }
32 });
33 }
34 }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
35 ee.setTimeout(task);
36 }
37  
38 protected RFuture<Boolean> renewExpirationAsync(long threadId) {
39 return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
40 "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
41 "redis.call('pexpire', KEYS[1], ARGV[1]); " +
42 "return 1; " +
43 "end; " +
44 "return 0;",
45 Collections.singletonList(getRawName()),
46 internalLockLeaseTime, getLockName(threadId));
47 }

  上面是两个方法的代码，一个定时任务Java代码，一个就是锁赎命的脚本代码，定时任务隔多久执行一次呢？看这段代码/ 3, TimeUnit.MILLISECONDS，实际就是internalLockLeaseTime这个变量，该变量的值来自于private long lockWatchdogTimeout = 30 * 1000，也就是10s。最后还是简单说下，赎命的脚本逻辑，hexists命令判断锁资源是不是自己的，如果是重新设置过期时间30s。好了以上就是redis实现分布式锁的逻辑，是不是比较麻烦？以上方案是不是万无一失了？确实也差不多了，但是仔细考虑还是有点问题的，这就是高并发系统的复杂之处。生产环境Redis至少是高可用集群，由于Redis本身实现的是AP方案，那么又有问题了，如果Master宕机了，Slave节点还没来得及同步数据，这时Slave节点做了Master，这时候锁资源丢失了，要解决这个问题好像挺难，当然Redis官方的红锁Redlock方案能解决这个问题，其实该方案的设计思路就cp原则，类似zookeeper的实现原理。   Redis和数据库双写一致性的问题：其实这也是一个比较麻烦的问题，以前浏览博客发现有朋友提出延迟双删的方案，并且还说是双保险的靠谱方案，这里我想说，老大一个BUG，逻辑都没通。我们简单看下问题的本质，其实还是上面那个问题，高并发情况下，数据同步问题，又是锁？是的，分布式锁？差不多，只是这里叫读写锁，读写锁我相信大部分平台都有实现，.NET、Java等等，原理跟数据库Repeatable read或者Serializable级别的控制差不多，写锁是独占锁，读锁是共享锁。这里我们还是看下redisson框架的实现。看代码：  

"local mode = redis.call('hget', KEYS[1], 'mode'); " +
"if (mode == false) then " +
"redis.call('hset', KEYS[1], 'mode', 'write'); " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (mode == 'write') then " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"local currentExpire = redis.call('pttl', KEYS[1]); " +
"redis.call('pexpire', KEYS[1], currentExpire + ARGV[1]); " +
"return nil; " +
"end; " +
"end;" +
"return redis.call('pttl', KEYS[1]);",

上面是获取写锁逻辑，代码就不详细解释了，还是基于Hash结构，只不过这里的读写锁是通过Mode字段实现，read表示读锁，write表示写锁，同样也是可重入锁。代码逻辑就到这吧。下面我们分析一下思路，如果现在有3个线程在操作key为a的值，1、3为读，2为写，123同样表示顺序，1在读取a时，获取了读锁，此时假如1号线程没有释放锁，2号写锁是加不了锁的，类似于数据库的s和ix、x锁不兼容，加入此时1号释放锁，2号拿到锁并且为写锁，3号是读取不了数据的，因为2号是独占锁，这也就就完美解决了一致性问题，但是性能确实不乐观，当然如果写少的情况下还是可以的。除了上面这种方案，还有一种同步的方案，通过三方服务，获取数据库操作日志，同步到Redis，比如阿里的开源框架canal，个人觉得一致性还是有蛮大问题，因为它们终究还是在网络环境里面，网络是个最不靠谱的东西。双写一致性就到这吧，最后简单聊下，redis的数据结构的应用。就到这吧，应用部分，后面再补上。