利用Redis设计库存系统的苦与乐
在秒杀等高并发场景下，既要保证库存安全，也要拥有极高的系统性能。从存储结构上，很多同学会选用Redis，毕竟Redis的单线程操作特性，很好地避免了线程安全的问题，同时具备极高的读写性能。
我们先来看下库存系统设计的几大核心要点：
1. 库存安全：既要保证线程安全，也要防止出现超卖
2. 同步响应：业务场景基本不允许异步响应库存扣减结果
3. 性能极限：在seckill场景下，性能总是被要求越高越好
我们来看下如何利用Redis来解决上面的三个问题。
一.库存安全
利用Redis来做库存扣减，避免超限的"方法"很多，坑也很多，我们先来看下常用的陷阱有哪些。
1. 先获取当前库存值进行比较，再进行扣减
defdecr_stock():conn=redis_conn()key="productA"current_storage=conn.get(key)current_storage_int=int(current_storage)ifcurrent_storage_int<=0 :return0result=conn.decr(key)returnresult
我们先在Redis中拿到当前的库存值，然后check是否已经扣减到了零，如果已经扣减到了零，则直接return；否则，就利用Redis的decr原子操作进行扣减，同时返回扣减后的库存值。
这种方法的问题很明显，在并发条件下，会出现脏读，设想一个场景，AB两个请求进来，A获取的库存值为1，B获取的库存值为1，然后两个请求都被发到redis中进行扣减操作，然后这种场景下，A最后得到的库存值为0；但是B最后得到的库存值为-1，超限。
2. 先扣减库存，再做比较，跟进情况是否做回滚
defdecr_stock():conn=redis_conn()key="productA"current=conn.decr(key)ifcurrent>=0:returncurrentelse: #回滚库存conn.incr(key)return0
直接先对库存值进行扣减，得到当前的库存值；然后，对此库存值进行check，如果库存>=0，则返回库存值，如果库存<0，则回滚库存，以便于防止负库存量的存在。
Redis Decr命令：DECR 命令会返回键 key 在执行减1操作之后的值。
这种做法引入了两个新的问题：
1).如果大批量的并发请求过来，redis承受的写操作的量，是加倍的，因为回滚库存的存在导致的。所以这种情况下，高并发量进来，极有可能将redis的写操作打出极限值，然后会出现很多redis写失败的错误警告
2). Redis的Decr操作和回滚操作无法保证原子性，在宕机情况下，容易产生数据不一致
3.先扣库存，然后通过整数溢出控制，根据情况进行回滚
defdecr_stock():conn=redis_conn()key="productA"current=conn.decr(key) #通过整数控制溢出的做法ifcheck_overflow(current):returncurrentelse: #回滚库存conn.incr(key)return0 defcheck_overflow(stock): #如果当前库存未被递减到0，则check_number为int类型，isinstance方法检测结果为true #如果当前库存已被递减到负数，则check_number为long类型，isinstance方法检测结果为falsecheck_number=sys.maxint - stockcheck_result=isinstance(check_number,int)returncheck_result
这种做法和方法2类似，只是比对部分由直接和0比对，变成了通过检测integer是否溢出的方式来进行。这样就彻底解决了高并发情况下，直接和零比对，限制不住的问题了。
虽然此种做法，相对于做法二说来，要靠谱很多，但是仍然解决不了在高并发情况下，redis写并发量加倍的问题，极有可能某个促销活动，在开始的那一刻，直接将redis的写操作打出问题来。
4.基于分布式锁的库存扣减
defdecr_stock():key ="productA" lock = getLock(key)iflocked ==1: current_storage = conn.get(key) current_storage_int = int(current_storage)ifcurrent_storage_int<=0:return0 result = conn.decr(key)returnresultelse:return"someone in it"
Redis在2.8以后支持Lua脚本的原子性操作，可以用来做分布式锁，解决超限的问题。
5. All in Lua
defstorage_scenario_six(): conn = redis_conn()lua =""" local storage = redis.call('get','storage_seckill') if storage ~= false then if tonumber(storage) > 0 then return redis.call('decr','storage_seckill') else return 'storage is zero now, can't perform decr action' end else return redis.call('set','storage_seckill',10) end """result = conn.eval(lua,0) print(result)
二、同步响应
如果只用Redis来进行存储，处理完数据直接返回前端即可。如果还要持久化到DB，要尽量避免直接操作DB，因为DB往往是最大的IO瓶颈，如果要异步落库到DB，比如使用MQ。要注意处理Redis扣减和消息发送的原子性处理。
三、性能
官网上redis的读写性能能到10W/QPS左右，这个量级应该可以解决绝大部分的场景。
但是经常有同学在压测的时候达不到这个性能，主要还是卡在网络环境上，在5W/QPS的时候，带宽就超过10M/s了。所有想追求Redis的极致性能，最好还是在同机房进行调用。
