本文是秒殺系統的第四篇，我們來討論秒殺系統中緩存熱點數據的問題，進一步延伸到數據庫和緩存的雙寫一致性問題。

在秒殺實際的業務中，一定有很多需要做緩存的場景，比如售賣的商品，包括名稱，詳情等。訪問量很大的數據，可以算是“熱點”數據了，尤其是一些讀取量遠大于寫入量的數據，更應該被緩存，而不應該讓請求打到數據庫上。

當然并不是所有的數據都需要進行緩存，那么一般哪些數據適合緩存呢？緩存量大但又不常變化的數據，比如詳情，評論等適合緩存。對于那些經常變化的數據，其實并不適合緩存，一方面會增加系統的復雜性（緩存的更新，緩存臟數據），另一方面也給系統帶來一定的不穩定性（緩存系統的維護）。上緩存之后，可以給我們帶來一定的好處：

• 能夠縮短服務的響應時間，給用戶帶來更好的體驗；
• 能夠增大系統的吞吐量，依然能夠提升用戶體驗；
• 減輕數據庫的壓力，防止高峰期數據庫被壓垮，導致整個線上服務OOM。

但是上了緩存，也會引入很多額外的問題：

• 緩存有多種選型，是內存緩存，memcached還是redis，你是否都熟悉，如果不熟悉，無疑增加了維護的難度（本來是個純潔的數據庫系統）；
• 緩存系統也要考慮分布式，比如redis的分布式緩存還會有很多坑，無疑增加了系統的復雜性；
• 在特殊場景下，如果對緩存的準確性有非常高的要求，就必須考慮緩存和數據庫的一致性問題。

本文想要重點討論的，就是緩存和數據庫的一致性問題。

緩存和數據庫雙寫一致性

在讀取緩存方面，大家沒啥疑問，都是按照下圖的流程來進行業務操作：

但是在更新緩存方面，對于更新完數據庫，再更新緩存呢，還是刪除緩存。又或者是先刪除緩存，再更新數據庫，其實大家存在很大的爭議。從理論上來說，給緩存設置過期時間，是保證最終一致性的解決方案。這種方案下，我們可以對存入緩存的數據設置過期時間，所有的寫操作以數據庫為準，對緩存操作只是盡最大努力即可。也就是說如果數據庫寫成功，緩存更新失敗，那么只要到達過期時間，則后面的讀請求自然會從數據庫中讀取新值然后回填緩存。因此，接下來討論的思路不依賴于給緩存設置過期時間這個方案。我們討論主要討論以下三種更新策略：

• 先更新數據庫，再更新緩存
• 刪除緩存，再更新數據庫
• 先更新數據庫，再刪除緩存

先更新數據庫，再更新緩存

這套方案，大家是普遍反對的。為什么呢？我們從以下兩個方面來進行說明。

線程安全

假設同時有請求A和請求B進行更新操作，那么會出現以下情況：

線程A更新了數據庫

線程B更新了數據庫

線程B更新了緩存

線程A更新了緩存

請求A更新緩存應該比請求B更新緩存早才對，但是因為網絡等原因，B卻比A更早更新了緩存。這就導致了臟數據，因此不考慮。

業務場景

如果是寫數據庫比較多，而讀數據比較少的業務需求，采用這種方案就會導致數據壓根還沒讀到，緩存就被頻繁的更新，浪費性能；

如果寫入數據庫的值，并不是直接寫入緩存的，而是要經過一系列復雜的計算再寫入緩存，那么每次寫入數據庫后，都再次計算寫入緩存的值，無疑是浪費性能的。顯然，刪除緩存更為適合。

刪除緩存，再更新數據庫

假設同時有一個請求A進行更新操作，另一個請求B進行查詢操作。那么會出現如下情形:

請求A進行寫操作，刪除緩存

請求B查詢發現緩存不存在

請求B去數據庫查詢得到舊值

請求B將舊值寫入緩存

請求A將新值寫入數據庫

上述情況會導致數據不一致的情形。如果不采用給緩存設置過期時間策略，該數據永遠都是臟數據。這種情況下我們就可以采用延時雙刪策略：先淘汰緩存，再寫數據庫最后再休眠1秒，再次淘汰緩存。當然這個休眠的時間，讀者應該自行評估自己的項目的讀數據業務邏輯的耗時，然后寫數據的休眠時間則在讀數據業務邏輯的耗時基礎上，加幾百ms即可。這么做的目的，就是確保讀請求結束，寫請求可以刪除讀請求造成的緩存臟數據。
有的人就會想到，如果我使用的是mysql的讀寫分離架構怎么辦？在這種情況下，造成數據不一致的原因如下，還是兩個請求，一個請求A進行更新操作，另一個請求B進行查詢操作：

請求A進行寫操作，刪除緩存

請求A將數據寫入數據庫了

請求B查詢緩存發現，緩存沒有值

請求B去從庫查詢，這時，還沒有完成主從同步，因此查詢到的是舊值

請求B將舊值寫入緩存

數據庫完成主從同步，從庫變為新值

這種情況還是使用雙刪延時策略。只是睡眠時間修改為在主從同步的延時時間基礎上，加幾百ms。那如果采用這種同步淘汰策略，吞吐量降低怎么辦？我們可以將第二次刪除作為異步的，自己起一個線程，異步刪除。這樣，寫的請求就不用沉睡一段時間后再返回，加大吞吐量。那如果第二次刪除刪除失敗怎么辦？第二次刪除失敗，就會出現如下情形。還是有兩個請求，一個請求A進行更新操作，另一個請求B進行查詢操作，為了方便，假設是單庫：

請求A進行寫操作，刪除緩存

請求B查詢發現緩存不存在

請求B去數據庫查詢得到舊值

請求B將舊值寫入緩存

請求A將新值寫入數據庫

請求A試圖去刪除請求B寫入對緩存值，結果失敗了。

如果第二次刪除緩存失敗，會再次出現緩存和數據庫不一致的問題。那么如何解決呢？請你繼續往下看。

先更新數據庫，再刪除緩存

假設這會有兩個請求，一個請求A做查詢操作，一個請求B做更新操作，那么會有如下情形產生：

緩存剛好失效

請求A查詢數據庫，得一個舊值

請求B將新值寫入數據庫

請求B刪除緩存

請求A將查到的舊值寫入緩存

如果發生上述情況，確實是會發生臟數據。但是發生這樣的情況的條件是這樣的：步驟3的寫數據庫操作比步驟2的讀數據庫操作耗時更短，才有可能使得步驟4先于步驟5。但是實際上數據庫的讀操作的速度遠快于寫操作的，因此步驟3耗時比步驟2更短，這一情形很難出現。那如果真的出現了怎么辦呢？首先，給緩存設有效時間是一種方案。其次，采用先刪除緩存，再更新數據庫策略里給出的異步延時刪除策略，保證讀請求完成以后，再進行刪除操作。這樣又回到上一個策略中遺留的問題：第二次刪除緩存失敗怎么辦？提供一個保障的重試機制即可，這里給出兩套方案。

方案一

如上圖，我們簡化一下步驟：

更新數據庫數據；

緩存因為種種問題刪除失敗

將需要刪除的key發送至消息隊列

自己消費消息，獲得需要刪除的key

繼續重試刪除操作，直到成功

該方案有一個缺點，對業務線代碼造成大量的侵入。于是有了方案二，在方案二中，啟動一個訂閱程序去訂閱數據庫的binlog，獲得需要操作的數據。在應用程序中，另起一段程序，獲得這個訂閱程序傳來的信息，進行刪除緩存操作。

方案二

如上圖，我們簡化一下步驟：

更新數據庫數據

數據庫會將操作信息寫入binlog日志當中

訂閱程序提取出所需要的數據以及key

另起一段非業務代碼，獲得該信息

嘗試刪除緩存操作，發現刪除失敗

將這些信息發送至消息隊列

重新從消息隊列中獲得該數據，重試操作。

讀取binlog的中間件，可以采用阿里開源的canal。到這里我們已經把緩存雙寫一致性的思路徹底梳理了一遍，下面對這幾種思路在我們原來代碼的基礎上進行代碼實戰，方便有需要的朋友參考。

秒殺實戰

先刪除緩存，再更新數據庫

我們在秒殺項目的代碼上OrderController中增加接口：先刪除緩存，再更新數據庫：

/*** 下單接口：先刪除緩存，再更新數據庫* @param sid* @return*/ @RequestMapping("/createOrderWithCacheV1/{sid}") @ResponseBody public String createOrderWithCacheV1(@PathVariable int sid) {int count = 0;try {// 刪除庫存緩存stockService.delStockCountCache(sid);// 完成扣庫存下單事務orderService.createPessimisticOrder(sid);} catch (Exception e) {LOGGER.error("購買失敗：[{}]", e.getMessage());return "購買失敗，庫存不足";}LOGGER.info("購買成功，剩余庫存為: [{}]", count);return String.format("購買成功，剩余庫存為：%d", count); }

stockService中新增：

@Override public void delStockCountCache(int id) {String hashKey = CacheKey.STOCK_COUNT.getKey() + "_" + id;stringRedisTemplate.delete(hashKey);LOGGER.info("刪除商品id：[{}] 緩存", id); }

先更新數據庫，再刪緩存

如果是先更新數據庫，再刪緩存，那么代碼只是在業務順序上顛倒了一下：

/*** 下單接口：先更新數據庫，再刪緩存* @param sid* @return*/ @RequestMapping("/createOrderWithCacheV2/{sid}") @ResponseBody public String createOrderWithCacheV2(@PathVariable int sid) {int count = 0;try {// 完成扣庫存下單事務orderService.createPessimisticOrder(sid);// 刪除庫存緩存stockService.delStockCountCache(sid);} catch (Exception e) {LOGGER.error("購買失敗：[{}]", e.getMessage());return "購買失敗，庫存不足";}LOGGER.info("購買成功，剩余庫存為: [{}]", count);return String.format("購買成功，剩余庫存為：%d", count); }

緩存延時雙刪

如何做延時雙刪呢，最好的方法是開設一個線程池，在線程中刪除key。更新前先刪除緩存，然后更新數據，再延時刪除緩存。OrderController中新增接口：

// 延時時間：預估讀數據庫數據業務邏輯的耗時，用來做緩存再刪除 private static final int DELAY_MILLSECONDS = 1000;/*** 下單接口：先刪除緩存，再更新數據庫，緩存延時雙刪* @param sid* @return*/ @RequestMapping("/createOrderWithCacheV3/{sid}") @ResponseBody public String createOrderWithCacheV3(@PathVariable int sid) {int count;try {// 刪除庫存緩存stockService.delStockCountCache(sid);// 完成扣庫存下單事務count = orderService.createPessimisticOrder(sid);// 延時指定時間后再次刪除緩存cachedThreadPool.execute(new delCacheByThread(sid));} catch (Exception e) {LOGGER.error("購買失敗：[{}]", e.getMessage());return "購買失敗，庫存不足";}LOGGER.info("購買成功，剩余庫存為: [{}]", count);return String.format("購買成功，剩余庫存為：%d", count); }

OrderController中新增線程池：

// 延時雙刪線程池 private static ExecutorService cachedThreadPool = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());/*** 緩存再刪除線程*/ private class delCacheByThread implements Runnable {private int sid;public delCacheByThread(int sid) {this.sid = sid;}public void run() {try {LOGGER.info("異步執行緩存再刪除，商品id：[{}]， 首先休眠：[{}] 毫秒", sid, DELAY_MILLSECONDS);Thread.sleep(DELAY_MILLSECONDS);stockService.delStockCountCache(sid);LOGGER.info("再次刪除商品id：[{}] 緩存", sid);} catch (Exception e) {LOGGER.error("delCacheByThread執行出錯", e);}} }

調用接口createOrderWithCacheV3

刪除緩存前庫存為48

刪除緩存前庫存為null,沒有數據

然后正常下單以后庫存變為47，此時將緩存更新到redis中

最后異步將緩存數據再次刪除：

的確是做了兩次緩存刪除：

刪除緩存重試機制

以上刪除有可能會失敗。要解決刪除失敗的問題，需要用到消息隊列，進行刪除操作的重試。這里我們為了達到效果，接入了RabbitMq，并且需要在接口中寫發送消息，并且需要消費者常駐來消費消息。
首先在pom.xml新增RabbitMq的依賴：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId> </dependency>

寫一個RabbitMqConfig：

@Configuration public class RabbitMqConfig {@Beanpublic Queue delCacheQueue() {return new Queue("delCache");} }

添加一個消費者：

@Component @RabbitListener(queues = "delCache") public class DelCacheReceiver {private static final Logger LOGGER = LoggerFactory.getLogger(DelCacheReceiver.class);@Autowiredprivate StockService stockService;@RabbitHandlerpublic void process(String message) {LOGGER.info("DelCacheReceiver收到消息: " + message);LOGGER.info("DelCacheReceiver開始刪除緩存: " + message);stockService.delStockCountCache(Integer.parseInt(message));} }

OrderController中新增接口：

/*** 下單接口：先更新數據庫，再刪緩存，刪除緩存重試機制* @param sid* @return*/ @RequestMapping("/createOrderWithCacheV4/{sid}") @ResponseBody public String createOrderWithCacheV4(@PathVariable int sid) {int count;try {// 完成扣庫存下單事務count = orderService.createPessimisticOrder(sid);// 刪除庫存緩存stockService.delStockCountCache(sid);// 延時指定時間后再次刪除緩存// cachedThreadPool.execute(new delCacheByThread(sid));// 假設上述再次刪除緩存沒成功，通知消息隊列進行刪除緩存sendDelCache(String.valueOf(sid));} catch (Exception e) {LOGGER.error("購買失敗：[{}]", e.getMessage());return "購買失敗，庫存不足";}LOGGER.info("購買成功，剩余庫存為: [{}]", count);return String.format("購買成功，剩余庫存為：%d", count); }

調用接口createOrderWithCacheV4

可以看到，我們先完成了下單，然后刪除了緩存，并且假設延遲刪除緩存失敗了，發送給消息隊列重試的消息，消息隊列收到消息后再去刪除緩存。

讀取binlog異步刪除緩存

這里我們使用阿里開源的canal來讀取binlog進行緩存的異步刪除。Canal用途很廣，并且上手非常簡單，我們在下一篇單獨做一下介紹。

猜你感興趣：
教你從0到1搭建秒殺系統-防超賣
教你從0到1搭建秒殺系統-限流
教你從0到1搭建秒殺系統-搶購接口隱藏與單用戶限制頻率
教你從0到1搭建秒殺系統-緩存與數據庫雙寫一致
教你從0到1搭建秒殺系統-Canal快速入門（番外篇）
教你從0到1搭建秒殺系統-訂單異步處理

更多文章請點擊：更多…

參考文章：
https://cloud.tencent.com/developer/article/1574827
https://www.jianshu.com/p/2936a5c65e6b
https://www.cnblogs.com/rjzheng/p/9041659.html
https://www.cnblogs.com/codeon/p/8287563.html
https://www.jianshu.com/p/0275ecca2438
https://www.jianshu.com/p/dc1e5091a0d8
https://coolshell.cn/articles/17416.html

總結

以上是生活随笔為你收集整理的教你从0到1搭建秒杀系统-缓存与数据库双写一致的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。