Redis實戰—秒殺優化（Redis消息隊列）

回顧

????????我們回顧一下前文下單的流程，當用戶發起請求，此時會請求nginx，nginx會訪問到tomcat，而tomcat中的程序，會進行串行操作，分成如下幾個步驟。
????????1、查詢優惠卷
????????2、判斷秒殺庫存是否足夠
????????3、查詢訂單
????????4、校驗是否滿足一人一單
????????5、扣減庫存
????????6、創建訂單
????????在這六步操作中，有很多操作是要去操作數據庫的，而且還是一個線程串行執行，這樣就會導致我們的程序執行的很慢，所以我們需要異步程序執行，那么該如何優化呢？

基于阻塞隊列實現程序異步優化

????????優化方案：我們將耗時較短的邏輯判斷交給redis操作，比如庫存是否足夠，是否滿足一人一單的條件，只要滿足這兩項判斷，就意味著我們一定可以完成下單，因此，我們只需要快速進行邏輯判斷，根本不用等下單邏輯走完，就可以先給用戶發送響應信息。若用戶可以下單，再在后臺開一個線程，讓后臺線程慢慢去執行queue里邊的消息，這樣程序耗時將大大縮短，而且也不用擔心線程池消耗殆盡的問題，因為我們的程序中并沒有手動使用任何線程池。

? ? ? ? 在用戶下單之后，判斷庫存是否充足只需要到redis中找到對應優惠券key的value（后臺在添加秒殺優惠券時，會對應將該優惠券及庫存添加到redis中），并判斷其是否大于0即可。如果不充足，則直接結束；如果充足，則繼續在redis中判斷該用戶是否已經下過單，如果set集合中不存在該用戶的記錄，則說明該用戶從未下過單，滿足下單條件。整個過程需要保證原子性，我們可以使用lua來操作。

優化代碼

?1. 首先，在添加優惠券的同時，我們需要將該優惠券及其庫存保存到redis中，方便我們之后在redis中快速判斷優惠券庫存是否充足。對添加優惠券方法做修改如下。

@Override
@Transactional
public void addSeckillVoucher(Voucher voucher) {// MP保存優惠券save(voucher);// 保存秒殺信息SeckillVoucher seckillVoucher = new SeckillVoucher();seckillVoucher.setVoucherId(voucher.getId());seckillVoucher.setStock(voucher.getStock());seckillVoucher.setBeginTime(voucher.getBeginTime());seckillVoucher.setEndTime(voucher.getEndTime());seckillVoucherService.save(seckillVoucher);// 保存秒殺庫存到Redis中stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY + voucher.getId(), voucher.getStock().toString());
}

?2. redis判斷采用lua腳本，代碼如下。

--1 參數列表
--1.1 優惠券id
local voucherId = ARGV[1]
--1.2 用戶id
local userId = ARGV[2]--2 數組key
--2.1 庫存key
local stockKey = 'seckill:stock:' .. voucherId
--2.2 訂單key
local orderKey = 'seckill:order:' .. voucherId--3 腳本業務
--3.1 判斷庫存是否充足
if (tonumber(redis.call('get', stockKey)) <= 0) then--3.2 若庫存不足，則返回1return 1
end--3.3 判斷用戶是否下單
if(redis.call('sismember',orderKey,userId)==1) then-- 3.4 存在，說明是重復下單，返回2return 2
end--3.5 扣庫存
redis.call('incrby', stockKey, -1)
--3.6 下單（保存用戶下單記錄）
redis.call('sadd', orderKey,userId)
return 0

3.?基于阻塞隊列實現秒殺優化

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate RedisIdWorker redisIdWorker;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));//初始化返回值SECKILL_SCRIPT.setResultType(Long.class);}@PostConstruct //注解含義：在當前類初始化完畢后執行private void init(){SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}//創建阻塞隊列private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024*1024);//創建線程池private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();//創建線程任務private class VoucherOrderHandler implements Runnable{@Overridepublic void run() {while (true){try {// 1.獲取隊列中的訂單信息VoucherOrder voucherOrder = orderTasks.take();// 2.創建訂單handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("處理訂單異常",e);}}}}private void handleVoucherOrder(VoucherOrder voucherOrder) {// 1.獲取用戶Long userId = voucherOrder.getUserId();// 2.創建鎖對象RLock lock = redissonClient.getLock("lock:order:" + userId);// 3.獲取鎖boolean isLock = lock.tryLock();// 4.若獲取鎖失敗if (!isLock) {log.error("不允許重復下單");return;}// 獲取鎖成功 （理論上沒有問題，lua腳本已經判斷過了，這里再加鎖只是兜底）try {//通過代理對象調用proxy.createVoucherOrder(voucherOrder);} finally {lock.unlock();}}@Transactionalpublic void createVoucherOrder(VoucherOrder voucherOrder) {boolean success = seckillVoucherService.update().setSql("stock = stock - 1")  //使用MP，設置sql語句.eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();save(voucherOrder);}private IVoucherOrderService proxy;@Overridepublic Result seckillVoucher(Long voucherId) {// 獲取用戶Long userId = UserHolder.getUser().getId();// 1.執行lua腳本，判斷用戶是否用購買資格（庫存不足與重復下單問題）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString());// 2.判斷結果是否為0int r = result.intValue();if(r!=0){// 2.1.不為0，代表沒有購物資格return Result.fail(r==1?"庫存不足":"不能重復下單");}// 2.2 為0，有購買資格，先創建訂單，再將訂單信息添加到阻塞隊列VoucherOrder voucherOrder = new VoucherOrder();// 2.3 獲取訂單id（Redis全局唯一id）long orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);voucherOrder.setUserId(userId);voucherOrder.setVoucherId(voucherId);// 2.4將訂單信息存入阻塞隊列，任務結束orderTasks.add(voucherOrder);//3.獲取代理對象，方便后序線程使用，可以放在成員變量或者是voucherOrder里面proxy = (IVoucherOrderService) AopContext.currentProxy();// 4.返回訂單idreturn Result.ok(orderId);}}

總結

秒殺業務的優化思路是什么？
1.先利用Redis完成庫存余量、一人一單判斷，創建搶單業務
2.再將搶單業務放入阻塞隊列，利用獨立線程異步下單

基于阻塞隊列的異步秒殺存在哪些問題？
1.內存限制問題
2.數據安全問題

Redis消息隊列

初始消息隊列

消息隊列就是存放消息的隊列，最簡單的消息隊列模型包括以下3個角色。
1.消息隊列：存儲和管理消息，也被稱為消息代理（Message Broker）
2.生產者：發送消息到消息隊列
3.消費者：從消息隊列獲取消息并處理消息

使用消息隊列的好處
? ? ? ? 消息隊列能夠將我們的業務進行解耦，舉一個生活中的例子就是：快遞員(生產者)把快遞放到快遞柜(Message Queue)里邊，我們(消費者)再從快遞柜里拿東西，這就是一個異步。如果耦合，相當于快遞員直接把快遞交給你，這事固然好，但是萬一你不在家，那么快遞員就會一直等你，這便浪費了快遞員的時間，所以這種思想在我們日常開發中，是非常有必要的。

基于List結構模擬消息隊列

????????消息隊列（Message Queue），字面意思就是存放消息的隊列。而Redis的list數據結構是一個雙向鏈表，很容易模擬出隊列效果。

????????隊列是入口和出口不在一邊，因此我們可以利用：LPUSH 結合 RPOP、或者 RPUSH 結合 LPOP來實現。不過要注意的是，當隊列中沒有消息時RPOP或LPOP操作會返回null，并不像JVM的阻塞隊列那樣會阻塞并等待消息。因此這里應該使用BRPOP或者BLPOP來實現阻塞效果。

基于List的消息隊列有哪些優缺點？

優點：
1.利用Redis存儲，不受限于JVM內存上限
2.基于Redis的持久化機制，數據安全性有保證
3.可以滿足消息有序性

缺點：
1.無法避免消息丟失
2.只支持單消費者

基于PubSub的消息隊列?

????????PubSub（發布訂閱）是Redis2.0版本引入的消息傳遞模型。顧名思義，消費者可以訂閱一個或多個channel，生產者向對應channel發送消息后，所有訂閱者都能收到相關消息。

SUBSCRIBE channel [channel] ：訂閱一個或多個頻道
PUBLISH channel msg ：向一個頻道發送消息
PSUBSCRIBE pattern[pattern] ：訂閱與pattern格式匹配的所有頻道
?

基于PubSub的消息隊列有哪些優缺點？

優點：
1.采用發布訂閱模型，支持多生產、多消費

缺點：
1.不支持數據持久化
2.無法避免消息丟失
3.消息堆積有上限，超出時數據丟失

基于Stream的消息隊列

Stream 是 Redis 5.0 引入的一種新數據類型，可以實現一個功能非常完善的消息隊列。

發送消息命令如下。

舉例如下。

讀取消息的方式之一：XREAD

舉例如下。
XREAD阻塞方式，讀取最新消息如下。
在業務開發中，我們可以循環的調用XREAD阻塞方式來查詢最新消息，從而實現持續監聽隊列的效果，偽代碼如下。
STREAM類型消息隊列的XREAD命令特點
1.消息可回溯
2.一個消息可以被多個消費者讀取
3.可以阻塞讀取
4.有消息漏讀的風險

基于Stream的消息隊列-消費者組

常用命令如下。?

XGROUP CREATE key groupName ID [MKSTREAM]
key：隊列名稱
groupName：消費者組名稱
ID：起始ID標示，$代表隊列中最后一個消息，0則代表隊列中第一個消息
MKSTREAM：隊列不存在時自動創建隊列刪除指定的消費者組
XGROUP DESTORY key groupName給指定的消費者組添加消費者
XGROUP CREATECONSUMER key groupname consumername刪除消費者組中的指定消費者
XGROUP DELCONSUMER key groupname consumername從消費者組讀取消息：
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
group：消費組名稱
consumer：消費者名稱，如果消費者不存在，會自動創建一個消費者
count：本次查詢的最大數量
BLOCK milliseconds：當沒有消息時最長等待時間
NOACK：無需手動ACK，獲取到消息后自動確認
STREAMS key：指定隊列名稱
ID：獲取消息的起始ID：">"：從下一個未消費的消息開始其它：根據指定id從pending-list中獲取已消費但未確認的消息。例如0，是從pending-list中的第一個消息開始

STREAM類型消息隊列的XREADGROUP命令特點
1.消息可回溯
2.可以多消費者爭搶消息，加快消費速度
3.可以阻塞讀取
4.沒有消息漏讀的風險
5.有消息確認機制，保證消息至少被消費一次

消費者監聽消息的基本思路如下圖所示。?

總結

代碼實現

基于Redis的Stream結構作為消息隊列，實現異步秒殺下單，需求如下。
1.創建一個Stream類型的消息隊列，名為stream.orders
2.修改之前的秒殺下單Lua腳本，在認定有搶購資格后，直接向stream.orders中添加消息，內容包3.含voucherId、userId、orderId
4.項目啟動時，開啟一個線程任務，嘗試獲取stream.orders中的消息，完成下單

首先創建消息隊列。

XGROUP CREATE stream.orders  g1 0 MKSTREAM

其次，修改Lua腳本如下。

--1 參數列表
--1.1 優惠券id
local voucherId = ARGV[1]
--1.2 用戶id
local userId = ARGV[2]
--1.3 訂單id
local orderId = ARGV[3]--2 數組key
--2.1 庫存key
local stockKey = 'seckill:stock:' .. voucherId
--2.2 訂單key
local orderKey = 'seckill:order:' .. voucherId--3 腳本業務
--3.1 判斷庫存是否充足
if (tonumber(redis.call('get', stockKey)) <= 0) then--3.2 若庫存不足，則返回1return 1
end--3.3 判斷用戶是否下單
if (redis.call('sismember', orderKey, userId) == 1) then-- 3.4 存在，說明是重復下單，返回2return 2
end--3.5 扣庫存
redis.call('incrby', stockKey, -1)
--3.6 下單（保存用戶下單記錄）
redis.call('sadd', orderKey, userId)
--3.7 發送消息到隊列中，XADD stream.order * k1 v1 k2 v2
redis.call('xadd', 'stream.orders','*','userId',userId,'voucherId',voucherId,'id',orderId)
return 0

改造秒殺業務代碼如下。

@Override
public Result seckillVoucher(Long voucherId) {// 獲取用戶Long userId = UserHolder.getUser().getId();// 獲取訂單id（Redis全局唯一id）long orderId = redisIdWorker.nextId("order");// 1.執行lua腳本，判斷用戶是否用購買資格（庫存不足與重復下單問題）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(), String.valueOf(orderId));// 2.判斷結果是否為0int r = result.intValue();if (r != 0) {// 2.1.不為0，代表沒有購物資格return Result.fail(r == 1 ? "庫存不足" : "不能重復下單");}// 3.先獲取代理對象，方便處理隊列訂單時使用proxy = (IVoucherOrderService) AopContext.currentProxy();// 4.返回訂單idreturn Result.ok(orderId);
}/*未修改前的代碼如下
@Override
public Result seckillVoucher(Long voucherId) {// 獲取用戶Long userId = UserHolder.getUser().getId();// 1.執行lua腳本，判斷用戶是否用購買資格（庫存不足與重復下單問題）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString());// 2.判斷結果是否為0int r = result.intValue();if(r!=0){// 2.1.不為0，代表沒有購物資格return Result.fail(r==1?"庫存不足":"不能重復下單");}// 2.2 為0，有購買資格，先創建訂單，再將訂單信息添加到阻塞隊列VoucherOrder voucherOrder = new VoucherOrder();// 2.3 獲取訂單id（Redis全局唯一id）long orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);voucherOrder.setUserId(userId);voucherOrder.setVoucherId(voucherId);// 2.4將訂單信息存入阻塞隊列，任務結束orderTasks.add(voucherOrder);// 3.先獲取代理對象，方便處理隊列訂單時使用proxy = (IVoucherOrderService) AopContext.currentProxy();// 4.返回訂單idreturn Result.ok(orderId);
}
*/

業務最終代碼如下。

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Resourceprivate ISeckillVoucherService seckillVoucherService;@Resourceprivate RedisIdWorker redisIdWorker;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Resourceprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));//初始化返回值SECKILL_SCRIPT.setResultType(Long.class);}@PostConstruct //注解含義：在當前類初始化完畢后執行private void init() {SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}//創建阻塞隊列private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);//創建線程池private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();//創建線程任務private class VoucherOrderHandler implements Runnable {String queueName = "stream.orders";@Overridepublic void run() {while (true) {try {// 1.獲取消息隊列中的訂單信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.ordersList<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create(queueName, ReadOffset.lastConsumed()));// 2.判斷消息獲取是否成功if (list == null || list.isEmpty()) {// 2.1.如果獲取失敗，說明沒有消息，繼續下一次循環continue;}// 3.解析消息中的訂單信息MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> values = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);// 3.如果獲取成功，可以下單handleVoucherOrder(voucherOrder);// 5.ACK確認 SACK stream.orders g1 idstringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());} catch (Exception e) {log.error("處理訂單異常", e);handlePendingList();}}}private void handlePendingList() {while (true) {try {// 1.獲取pending-list中的訂單信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.ordersList<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1),StreamOffset.create(queueName, ReadOffset.from("0")));// 2.判斷消息獲取是否成功if (list == null || list.isEmpty()) {// 2.1.如果獲取失敗，說明沒有消息，跳出循環break;}// 3.解析消息中的訂單信息MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> values = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);// 3.如果獲取成功，可以下單handleVoucherOrder(voucherOrder);// 5.ACK確認 SACK stream.orders g1 idstringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());handleVoucherOrder(voucherOrder);} catch (Exception e) {log.error("處理pending-list訂單異常", e);try {Thread.sleep(20);} catch (InterruptedException ex) {ex.printStackTrace();}}}}}private void handleVoucherOrder(VoucherOrder voucherOrder) {// 1.獲取用戶Long userId = voucherOrder.getUserId();// 2.創建鎖對象RLock lock = redissonClient.getLock("lock:order:" + userId);// 3.獲取鎖boolean isLock = lock.tryLock();// 4.若獲取鎖失敗if (!isLock) {log.error("不允許重復下單");return;}// 獲取鎖成功try {//通過代理對象調用proxy.createVoucherOrder(voucherOrder);} finally {lock.unlock();}}private IVoucherOrderService proxy;@Overridepublic Result seckillVoucher(Long voucherId) {// 獲取用戶Long userId = UserHolder.getUser().getId();// 獲取訂單id（Redis全局唯一id）long orderId = redisIdWorker.nextId("order");// 1.執行lua腳本，判斷用戶是否用購買資格（庫存不足與重復下單問題）Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(), String.valueOf(orderId));// 2.判斷結果是否為0int r = result.intValue();if (r != 0) {// 2.1.不為0，代表沒有購物資格return Result.fail(r == 1 ? "庫存不足" : "不能重復下單");}// 3.先獲取代理對象，方便處理隊列訂單時使用proxy = (IVoucherOrderService) AopContext.currentProxy();// 4.返回訂單idreturn Result.ok(orderId);}@Transactionalpublic void createVoucherOrder(VoucherOrder voucherOrder) {boolean success = seckillVoucherService.update().setSql("stock = stock - 1")  //使用MP，設置sql語句.eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();save(voucherOrder);}
}