技術派中的緩存一緻性解決方案

今天結合技術派項目，告訴大家如何去實作 MySQL 和 Redis 的一緻性。

在講解實戰部分之前，我們還是先回顧一下理論知識，根據網上的衆多解決方案，我們總結出 6 種：

你可以先想想，技術派會采用哪種方案呢？

理論知識

溫馨提示：如果你對理論知識已經非常清楚，可以直接跳到文章的實戰部分。

不好的方案

1. 先寫 MySQL，再寫 Redis

圖解說明：

這是一副時序圖，描述請求的先後調用順序；

橘黃色的線是請求 A，黑色的線是請求 B；

橘黃色的文字，是 MySQL 和 Redis 最終不一緻的資料；

資料是從 10 更新為 11；

後面所有的圖，都是這個含義，不再贅述。

請求 A、B 都是先寫 MySQL，然後再寫 Redis，在高并發情況下，如果請求 A 在寫 Redis 時卡了一會，請求 B 已經依次完成資料的更新，就會出現圖中的問題。

這個圖已經畫的很清晰了，我就不用再去啰嗦了吧，不過這裡有個前提，就是對于讀請求，先去讀 Redis，如果沒有，再去讀 DB，但是讀請求不會再回寫 Redis。 大白話說一下，就是讀請求不會更新 Redis。

2. 先寫 Redis，再寫 MySQL

同“先寫 MySQL，再寫 Redis”，看圖可秒懂。

3. 先删除 Redis，再寫 MySQL

這幅圖和上面有些不一樣，前面的請求 A 和 B 都是更新請求，這裡的請求 A 是更新請求，但是請求 B 是讀請求，且請求 B 的讀請求會回寫 Redis。

請求 A 先删除緩存，可能因為卡頓，資料一直沒有更新到 MySQL，導緻兩者資料不一緻。

這種情況出現的機率比較大，因為請求 A 更新 MySQL 可能耗時會比較長，而請求 B 的前兩步都是查詢，會非常快。

好的方案

4. 先删除 Redis，再寫 MySQL，再删除 Redis

對于“先删除 Redis，再寫 MySQL”，如果要解決最後的不一緻問題，其實再對 Redis 重新删除即可，這個也是大家常說的“緩存雙删”。

為了便于大家看圖，對于藍色的文字，“删除緩存 10”必須在“回寫緩存10”後面，那如何才能保證一定是在後面呢？網上給出的第一個方案是，讓請求 A 的最後一次删除，等待 500ms。

對于這種方案，看看就行，反正我是不會用，太 Low 了，風險也不可控。

那有沒有更好的方案呢，我建議異步串行化删除，即删除請求入隊列

異步删除對線上業務無影響，串行化處理保障并發情況下正确删除。

如果雙删失敗怎麼辦，網上有給 Redis 加一個緩存過期時間的方案，這個不敢苟同。個人建議整個重試機制，可以借助消息隊列的重試機制，也可以自己整個表，記錄重試次數，方法很多。

簡單小結一下：

“緩存雙删”不要用無腦的 sleep 500 ms；

通過消息隊列的異步&串行，實作最後一次緩存删除；

緩存删除失敗，增加重試機制。

5. 先寫 MySQL，再删除 Redis

對于上面這種情況，對于第一次查詢，請求 B 查詢的資料是 10，但是 MySQL 的資料是 11，隻存在這一次不一緻的情況，對于不是強一緻性要求的業務，可以容忍。（那什麼情況下不能容忍呢，比如秒殺業務、庫存服務等。）

當請求 B 進行第二次查詢時，因為沒有命中 Redis，會重新查一次 DB，然後再回寫到 Reids。

這裡需要滿足 2 個條件：

• 緩存剛好自動失效；
• 請求 B 從資料庫查出 10，回寫緩存的耗時，比請求 A 寫資料庫，并且删除緩存的還長。

對于第二個條件，我們都知道更新 DB 肯定比查詢耗時要長，是以出現這個情況的機率很小，同時滿足上述條件的情況更小。

6. 先寫 MySQL，通過 Binlog，異步更新 Redis

這種方案，主要是監聽 MySQL 的 Binlog，然後通過異步的方式，将資料更新到 Redis，這種方案有個前提，查詢的請求，不會回寫 Redis。

這個方案，會保證 MySQL 和 Redis 的最終一緻性，但是如果中途請求 B 需要查詢資料，如果緩存無資料，就直接查 DB；如果緩存有資料，查詢的資料也會存在不一緻的情況。

是以這個方案，是實作最終一緻性的終極解決方案，但是不能保證明時性。

幾種方案比較

我們對比上面讨論的 6 種方案：

1. 先寫 Redis，再寫 MySQL

• 這種方案，我肯定不會用，萬一 DB 挂了，你把資料寫到緩存，DB 無資料，這個是災難性的；
• 我之前也見同學這麼用過，如果寫 DB 失敗，對 Redis 進行逆操作，那如果逆操作失敗呢，是不是還要搞個重試？

1. 先寫 MySQL，再寫 Redis

• 對于并發量、一緻性要求不高的項目，很多就是這麼用的，我之前也經常這麼搞，但是不建議這麼做；
• 當 Redis 瞬間不可用的情況，需要報警出來，然後線下處理。

1. 先删除 Redis，再寫 MySQL

• 這種方式，我還真沒用過，直接忽略吧。

1. 先删除 Redis，再寫 MySQL，再删除 Redis

• 這種方式雖然可行，但是感覺好複雜，還要搞個消息隊列去異步删除 Redis。

1. 先寫 MySQL，再删除 Redis

• 比較推薦這種方式，删除 Redis 如果失敗，可以再多重試幾次，否則報警出來；
• 這個方案，是實時性中最好的方案，在一些高并發場景中，推薦這種。

1. 先寫 MySQL，通過 Binlog，異步更新 Redis

• 對于異地容災、資料彙總等，建議會用這種方式，比如 binlog + kafka，資料的一緻性也可以達到秒級；
• 純粹的高并發場景，不建議用這種方案，比如搶購、秒殺等。

個人結論：

• 實時一緻性方案：采用“先寫 MySQL，再删除 Redis”的政策，這種情況雖然也會存在兩者不一緻，但是需要滿足的條件有點苛刻，是以是滿足實時性條件下，能盡量滿足一緻性的最優解。
• 最終一緻性方案：采用“先寫 MySQL，通過 Binlog，異步更新 Redis”，可以通過 Binlog，結合消息隊列異步更新 Redis，是最終一緻性的最優解。

項目實戰

資料更新

因為項目對實時性要求高，是以采用方案 5，先寫 MySQL，再删除 Redis 的方式。

下面隻是一個示例，我們将文章的标簽放入 MySQL 之後，再删除 Redis，所有涉及到 DB 更新的操作都需要按照這種方式處理。

這裡加了一個事務，如果 Redis 删除失敗，MySQL 的更新操作也需要復原，避免查詢時讀取到髒資料。

@Override
@Transactional(rollbackFor = Exception.class)
public void saveTag(TagReq tagReq) {
    TagDO tagDO = ArticleConverter.toDO(tagReq);

    // 先寫 MySQL
    if (NumUtil.nullOrZero(tagReq.getTagId())) {
        tagDao.save(tagDO);
    } else {
        tagDO.setId(tagReq.getTagId());
        tagDao.updateById(tagDO);
    }

    // 再删除 Redis
    String redisKey = CACHE_TAG_PRE + tagDO.getId();
    RedisClient.del(redisKey);
}

@Override
@Transactional(rollbackFor = Exception.class)
public void deleteTag(Integer tagId) {
    TagDO tagDO = tagDao.getById(tagId);
    if (tagDO != null){
        // 先寫 MySQL
        tagDao.removeById(tagId);

        // 再删除 Redis
        String redisKey = CACHE_TAG_PRE + tagDO.getId();
        RedisClient.del(redisKey);
    }
}

@Override
public void operateTag(Integer tagId, Integer pushStatus) {
    TagDO tagDO = tagDao.getById(tagId);
    if (tagDO != null){

        // 先寫 MySQL
        tagDO.setStatus(pushStatus);
        tagDao.updateById(tagDO);

        // 再删除 Redis
        String redisKey = CACHE_TAG_PRE + tagDO.getId();
        RedisClient.del(redisKey);
    }
}
           

資料擷取

這個也很簡單，先查詢緩存，如果有就直接傳回；如果未查詢到，需要先查詢 DB ，再寫入緩存。

我們放入緩存時，加了一個過期時間，用于兜底，萬一兩者不一緻，緩存過期後，資料會重新更新到緩存。

@Override
public TagDTO getTagById(Long tagId) {

    String redisKey = CACHE_TAG_PRE + tagId;

    // 先查詢緩存，如果有就直接傳回
    String tagInfoStr = RedisClient.getStr(redisKey);
    if (tagInfoStr != null && !tagInfoStr.isEmpty()) {
        return JsonUtil.toObj(tagInfoStr, TagDTO.class);
    }

    // 如果未查詢到，需要先查詢 DB ，再寫入緩存
    TagDTO tagDTO = tagDao.selectById(tagId);
    tagInfoStr = JsonUtil.toStr(tagDTO);
    RedisClient.setStrWithExpire(redisKey, tagInfoStr, CACHE_TAG_EXPRIE_TIME);

    return tagDTO;
}
           

測試用例

/**
 * @author Louzai
 * @date 2023/5/5
 */
@Slf4j
public class MysqlRedisService extends BasicTest {

    @Autowired
    private TagSettingService tagSettingService;

    @Test
    public void save() {
        TagReq tagReq = new TagReq();
        tagReq.setTag("Java");
        tagReq.setTagId(1L);
        tagSettingService.saveTag(tagReq);
        log.info("save success:{}", tagReq);
    }

    @Test
    public void query() {
        TagDTO tagDTO = tagSettingService.getTagById(1L);
        log.info("query tagInfo:{}", tagDTO);
    }
}
           

我們看一下 Redis：

127.0.0.1:6379> get pai_cache_tag_pre_1
"{\"tagId\":1,\"tag\":\"Java\",\"status\":1,\"selected\":null}"
           

以及結果輸出：

後記

這篇文章很基礎，也非常适用，大家可以直接下載下傳技術派項目，裡面都有代碼和測試用例，代碼倉庫詳見：https://github.com/itwanger/paicoding