秒殺業務優化之從分布式鎖到基于消息隊列的異步秒殺

一、業務場景介紹

? ? ? ? 優惠券、門票等限時搶購常常出現在各類應用中，這樣的業務一般為了引流宣傳而降低利潤，所以一旦出現問題將造成較大損失，那么在業務中就要求我們對這類型商品嚴格限時、限量、每位用戶限一次、準確無誤的創建訂單，這樣的要求看似簡單，但在分布式系統中，要求我們充分考慮高并發下的線程安全問題，今天我們來看一下兩種解決思路。

二、基于Redisson分布式鎖的秒殺方案

????????這里我們就不進行自定義redis鎖了，Redisson 基于 Redis 實現了?Java 駐內存數據網格（In-Memory Data Grid），它不僅提供了對 Redis 原生命令的封裝，還提供了一系列高級的分布式數據結構和服務，促進使用者對 Redis 的關注分離，讓開發者能夠更專注于業務邏輯，所以我們直接使用Redisson，但底層源碼還是需要我們去自己學習掌握的。

1.流程概覽

????????其實單看流程圖我們就能發現這一連串的串行邏輯就會非常影響效率，我們先留著這個問題后面優化。

2.具體實現

    @Overridepublic Result generate(Long voucherId) {//查詢優惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);//活動是否開始/結束if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {return Result.fail("活動未開始！");}if (voucher.getEndTime().isBefore(LocalDateTime.now())) {return Result.fail("活動已結束！");}//庫存表是否充足if (voucher.getStock()<1) {return Result.fail("庫存不足！");}Long userId = UserHolder.getUser().getId();//只鎖同一個id//創建鎖對象RLock lock = redissonClient.getLock("lock:order:" + userId);//獲取鎖,防止同一用戶的并發請求boolean isLock = lock.tryLock();//默認不等待，30秒過期if (!isLock) {//獲取鎖失敗return Result.fail("網絡繁忙！");}//拿到spring事務代理，這里為了簡單解決事務自調用直接去拿代理可能造成問題，建議將事務方法重構至另一服務類并注入try {IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);} finally {//釋放鎖lock.unlock();}}@Transactional//要鎖住事物，防止事物在鎖釋放后才提交導致其他線程進入public Result createVoucherOrder(Long voucherId) {Long userId = UserHolder.getUser().getId();//一人一單int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();if (count >0) {return Result.fail("您最多只可購買一單！");}//扣減庫存boolean flag = seckillVoucherService.update().setSql("stock=stock-1").eq("voucher_id", voucherId).gt("stock",0).update();if (!flag){//高并發下已經被其他用戶線程扣減return Result.fail("庫存不足2！");}//創建訂單VoucherOrder voucherOrder = new VoucherOrder();//唯一IDlong orderId = redisIdWorker.nextId("order");voucherOrder.setId(orderId);//用戶idvoucherOrder.setUserId(userId);//代金券IdvoucherOrder.setVoucherId(voucherId);save(voucherOrder);//返回訂單IDreturn Result.ok(orderId);}

3.測試分析

?接下來我們登錄數據庫中所有的用戶并記錄Authorization

@SpringBootTest
@Component
public class SecKill {@Autowiredprivate IUserService userService;@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Testvoid userLogin() throws IOException {// 定義保存 token 的文件路徑String filePath = "D:\\tokens.txt";// 使用 BufferedWriter 寫入文件try (BufferedWriter writer = new BufferedWriter(new FileWriter(filePath, true))) { // 追加模式for (User user : userService.list()) {String phone = user.getPhone();HttpSession session = null;userService.sendCode(phone, session);String code = stringRedisTemplate.opsForValue().get("login:code:" + phone);LoginFormDTO loginFormDTO = new LoginFormDTO();loginFormDTO.setCode(code);loginFormDTO.setPhone(phone);String token = userService.logIn(loginFormDTO, session);// 將 token 寫入文件writer.write(token);writer.newLine(); // 換行writer.flush(); // 刷新緩沖區，確保數據寫入文件}} catch (IOException e) {e.printStackTrace();}}
}

然后我們設置優惠券數量為200，通過jmeter（一款測試工具，大家自行學習如何使用）模擬數據庫中1000多個用戶總計每秒1000的高并發請求

?從聚合報告中可以看到雖然80%的異常率確實滿足了我們對優惠券的限量要求，通過查看數據庫訂單和庫存也不存在問題，但是我們可以看到我們的平均響應時間在高并發下達到了344ms，吞吐量只有1200左右，如果面臨更高的并發難免因性能局限出現問題。

三、基于消息隊列的異步秒殺

1.問題分析

正如我們一開始發現的，每個請求來到服務器都需要執行一串的數據庫讀寫操作，而寫操作耗時是比較久的，可是當我們確定用戶搶單成功后只要能確保訂單最終寫入即可，無需讓其阻塞請求，所以我們其實可以將讀寫操作分離開。

我們可以利用讀操作完成下單資格的各種校驗，校驗成功即可對請求做出響應，那么后續寫訂單操作怎么完成呢？我們需要根據校驗成功的記錄完成寫操作，那誰來完成校驗成功的記錄呢，這樣記錄是不是又和原來的讀寫串行一樣了呢？

2.工具對比

首先我們的目的是加快請求響應效率，減輕數據庫壓力，其實我們需要的就是一個中間工具做到能夠快速存儲校驗成功的記錄并有限制的可控的逐漸將存儲起來的記錄轉發給數據庫讓其創建訂單，能做到上述要求的工具有很多，這里簡單對比以下三種供大家參考。

特性/技術	阻塞隊列	Redis	MQ消息中間件（如RabbitMQ、Kafka）
系統解耦	低，主要用于單機環境	中，支持集群部署	高，天然用于系統解耦
異步通信	支持，但需要手動實現	通過發布/訂閱模式實現	專為異步通信設計
削峰填谷	臨時存儲請求，能力有限	緩存請求，需合理設計策略	緩存大量請求，后端按速率消費
可靠性和持久性	依賴具體實現，需額外持久化	支持持久化，可靠性較高	高可靠性和持久性，支持消息確認
性能和吞吐量	受限于單機處理能力	性能較高，支持集群	最高，適用于大規模分布式系統
功能豐富性	單一，主要用于線程間通信	支持多種數據結構和操作	支持多種消息協議、路由機制等
開發和維護成本	低，但需手動實現異步邏輯	中等，易于實現和使用	高，需學習和理解相關協議和機制
適用場景	小規模、單機環境	中小規模、集群部署	大規模分布式系統、復雜路由

?3.流程概覽

????????由于阻塞隊列局限較大，MQ中間件比較簡單，這里我們以Redis中的stream為例（除此之外，list和PubSub也能實現，但是局限較大）實現異步秒殺。

對于紅框部分，為了確保原子性，我們借助lua腳本完成，這樣一來我們就將MySQL的讀寫操作分離開來，請求響應中只需要讀取驗證，用redis更高效的io操作完成簡單記錄，隨后異步逐漸處理MySQl的訂單寫入。

4.具體實現

lua腳本

---
--- Generated by EmmyLua(https://github.com/EmmyLua)
--- Created by cds.
--- DateTime: 2025/3/23 13:03
---
--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--腳本業務
--判斷庫存是否充足
if (tonumber(redis.call('get',stockKey))<=0) then--庫存不足返回1return 1
end
--判斷用戶是否下單
if (redis.call('sismember',orderKey,userId)==1) then--下過單返回2return 2
end
--扣庫存
redis.call('incrby',stockKey,-1)
--下單
redis.call('sadd',orderKey,userId)
--發送消息到消息隊列  xadd stream.orders * k1 v1 k2 v2 ..
redis.call('xadd','stream.orders','*','userId',userId,'voucherId',voucherId,"id",orderId)return 0

?具體業務

    @Autowiredprivate IVoucherOrderService proxy;//初始化lua腳本信息private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT =new DefaultRedisScript<>();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));SECKILL_SCRIPT.setResultType(Long.class);}//異步單例線程private static final ExecutorService SECKILL_ORDER_EXECTUOR= Executors.newSingleThreadExecutor();//在spring的Bean初始化并注入后開始@PostConstructprivate void init(){SECKILL_ORDER_EXECTUOR.submit(new VoucherOrderHandler());}//線程任務private class VoucherOrderHandler implements Runnable {String queueName = "stream.orders";@Overridepublic void run() {while (true) {try {//獲取消息隊列中的訂單信息  XREAD GROUP group1 c1 count 1 block 2000 streams stream.orders >List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("group1", "c1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create(queueName, ReadOffset.lastConsumed()));//判斷消息是否獲取成if (list == null || list.isEmpty()) {//獲取失敗 沒有消息，繼續循環continue;}//獲取成功，可以下單//解析消息中的訂單信息MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);handleVoucherOrder(voucherOrder);//ACK確認 SACK stream.orders group1 idstringRedisTemplate.opsForStream().acknowledge(queueName, "group1", record.getId());} catch (Exception e) {log.error("創建訂單異常{}", e.getMessage());//有異常去pendingList拿handlePendingList();}}}private void handlePendingList() {while (true) {try {//獲取pending-list隊列中的訂單信息  XREAD GROUP group1 c1 count 1 block 2000 streams stream.orders 0List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("group1", "c1"),StreamReadOptions.empty().count(1),StreamOffset.create(queueName, ReadOffset.from("0")));//判斷消息是否獲取成if (list == null || list.isEmpty()) {//獲取失敗 pending-list沒有消息，結束循環break;}//獲取成功，可以下單//解析消息中的訂單信息MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);handleVoucherOrder(voucherOrder);//ACK確認 SACK stream.orders group1 idstringRedisTemplate.opsForStream().acknowledge(queueName, "group1", record.getId());} catch (Exception e) {log.error("創建訂單異常{}", e.getMessage());try {Thread.sleep(20);} catch (InterruptedException ex) {throw new RuntimeException(ex);}}}}}private void handleVoucherOrder(VoucherOrder voucherOrder) {Long userId = voucherOrder.getUserId();//創建鎖對象RLock lock = redissonClient.getLock("lock:order:" + userId);//獲取鎖boolean isLock = lock.tryLock();//默認不等待，30秒過期if (!isLock) {//獲取鎖失敗log.info("請勿重復購買！");return;}//拿到spring事務代理try {proxy.createVoucherOrder(voucherOrder);} finally {//釋放鎖lock.unlock();}}//這部分的檢驗是以防stream消息隊列里出現問題導致重復save操作@Transactional//要鎖住事物public void createVoucherOrder(VoucherOrder voucherOrder) {Long userId = voucherOrder.getUserId();//一人一單int count = query().eq("user_id", userId).eq("voucher_id", voucherOrder.getVoucherId()).count();if (count > 0) {log.error("您最多只可購買一單！");return;}//扣減庫存boolean flag = seckillVoucherService.update().setSql("stock=stock-1").eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();if (!flag) {log.info("庫存不足！");return;}//創建訂單save(voucherOrder);}}@Overridepublic Result secKill(Long voucherId) {//查詢優惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);//活動是否開始/結束if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {return Result.fail("活動未開始！");}if (voucher.getEndTime().isBefore(LocalDateTime.now())) {return Result.fail("活動已結束！");}//庫存表是否充足if (voucher.getStock() < 1) {return Result.fail("庫存不足！");}//獲取用戶Long userId = UserHolder.getUser().getId();//1執行lua腳本//唯一IDlong orderId = redisIdWorker.nextId("order");Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), userId.toString(), String.valueOf(orderId));int r = result.intValue();//2判斷lua腳本返回值0if (r != 0) {//2.1不為零無資格return Result.fail(r == 1 ? "庫存不足！" : "不能重復下單！");}return Result.ok(orderId);}