前言

1. 需求

2. 性能優化

3. 出問題了

4. 多線程消費

5. 順序消費

6. 唯一索引

7. 分布式鎖

8. 統一mq異步處理

9. insert on duplicate key update

10. insert ignore

11. 防重表

前言

最近測試給我提了一個bug，說我之前提供的一個批量復制商品的接口，產生了重復的商品數據。

追查原因之后發現，這個事情沒想象中簡單，可以說一波多折。

1. 需求

產品有個需求：用戶選擇一些品牌，點擊確定按鈕之后，系統需要基于一份默認品牌的商品數據，復制出一批新的商品。

拿到這個需求時覺得太簡單了，三下五除二就搞定。

我提供了一個復制商品的基礎接口，給商城系統調用。

當時的流程圖如下：

如果每次復制的商品數量不多，使用同步接口調用的方案問題也不大。

2. 性能優化

但由于每次需要復制的商品數量比較多，可能有幾千。

如果每次都是用同步接口的方式復制商品，可能會有性能問題。

因此，后來我把復制商品的邏輯改成使用mq異步處理。

改造之后的流程圖：

復制商品的結果還需要通知商城系統：

這個方案看起來，挺不錯的。

但后來出現問題了。

3. 出問題了

測試給我們提了一個bug，說我之前提供的一個批量復制商品的接口，產生了重復的商品數據。

經過追查之后發現，商城系統為了性能考慮，也改成異步了。

他們沒有在接口中直接調用基礎系統的復制商品接口，而是在job中調用的。

站在他們的視角流程圖是這樣的：

用戶調用商城的接口，他們會往請求記錄表中寫入一條數據，然后在另外一個job中，異步調用基礎系統的接口去復制商品。

但實際情況是這樣的：商城系統內部出現了bug，在請求記錄表中，同一條請求產生了重復的數據。這樣導致的結果是，在job中調用基礎系統復制商品接口時，發送了重復的請求。

剛好基礎系統現在是使用RocketMQ異步處理的。由于商城的job一次會取一批數據（比如:20條記錄），在極短的時間內（其實就是在一個for循環中）多次調用接口，可能存在相同的請求參數連續調用復制商品接口情況。于是，出現了并發插入重復數據的問題。

為什么會出現這個問題呢？

4. 多線程消費

RocketMQ的消費者，為了性能考慮，默認是用多線程并發消費的，最大支持64個線程。

例如：

@RocketMQMessageListener(topic="${com.susan.topic:PRODUCT_TOPIC}",
consumerGroup="${com.susan.group:PRODUCT_TOPIC_GROUP}")
@Service
publicclassMessageReceiverimplementsRocketMQListener{

@Override
publicvoidonMessage(MessageExtmessage){
Stringmessage=newString(message.getBody(),StandardCharsets.UTF_8);
doSamething(message);
}
}

也就是說，如果在極短的時間內，連續發送重復的消息，就會被不同的線程消費。

即使在代碼中有這樣的判斷：

ProductoldProduct=query(hashCode);
if(oldProduct==null){
productMapper.insert(product);
}

在插入數據之前，先判斷該數據是否已經存在，只有不存在才會插入。

但由于在并發情況下，不同的線程都判斷商品數據不存在，于是同時進行了插入操作，所以就產生了重復數據。

如下圖所示：

5. 順序消費

為了解決上述并發消費重復消息的問題，我們從兩方面著手：

商城系統修復產生重復記錄的bug。

基礎系統將消息改成單線程順序消費。

我仔細思考了一下，如果只靠商城系統修復bug，以后很難避免不出現類似的重復商品問題，比如：如果用戶在極短的時間內點擊創建商品按鈕多次，或者商城系統主動發起重試。

所以，基礎系統還需進一步處理。

其實RocketMQ本身是支持順序消費的，需要消息的生產者和消費者一起改。

生產者改為：

rocketMQTemplate.asyncSendOrderly(topic,message,hashKey,newSendCallback(){
@Override
publicvoidonSuccess(SendResultsendResult){
log.info("sendMessagesuccess");
}

@Override
publicvoidonException(Throwablee){
log.error("sendMessagefailed!");
}
});

重點是要調用rocketMQTemplate對象的asyncSendOrderly方法，發送順序消息。

消費者改為：

@RocketMQMessageListener(topic="${com.susan.topic:PRODUCT_TOPIC}",
consumeMode=ConsumeMode.ORDERLY,
consumerGroup="${com.susan.group:PRODUCT_TOPIC_GROUP}")
@Service
publicclassMessageReceiverimplementsRocketMQListener{

@Override
publicvoidonMessage(MessageExtmessage){
Stringmessage=newString(message.getBody(),StandardCharsets.UTF_8);
doSamething(message);
}
}

接收消息的重點是RocketMQMessageListener注解中的consumeMode參數，要設置成ConsumeMode.ORDERLY，這樣就能順序消費消息了。

兩邊都修改之后，復制商品這一塊就沒有再出現重復商品的問題了。

But，修完bug之后，我又思考了良久。

復制商品只是創建商品的其中一個入口，如果有其他入口，跟復制商品功能同時創建新商品呢？

不也會出現重復商品問題？

雖說，這種概率非常非常小。

但如果一旦出現重復商品問題，后續涉及到要合并商品的數據，非常麻煩。

經過這一次的教訓，一定要防微杜漸。

不管是用戶，還是自己的內部系統，從不同的入口創建商品，都需要解決重復商品創建問題。

那么，如何解決這個問題呢？

6. 唯一索引

解決重復商品數據問題，最快成本最低最有效的辦法是：給表建唯一索引。

想法是好的，但我們這邊有個規范就是：業務表必須都是邏輯刪除。

而我們都知道，要刪除表的某條記錄的話，如果用delete語句操作的話。

例如：

deletefromproductwhereid=123;

這種delete操作是物理刪除，即該記錄被刪除之后，后續通過sql語句基本查不出來。（不過通過其他技術手段可以找回，那是后話了）

還有另外一種是邏輯刪除，主要是通過update語句操作的。

例如：

updateproductsetdelete_status=1,edit_time=now(3)
whereid=123;

邏輯刪除需要在表中額外增加一個刪除狀態字段，用于記錄數據是否被刪除。在所有的業務查詢的地方，都需要過濾掉已經刪除的數據。

通過這種方式刪除數據之后，數據任然還在表中，只是從邏輯上過濾了刪除狀態的數據而已。

其實對于這種邏輯刪除的表，是沒法加唯一索引的。

為什么呢？

假設之前給商品表中的name和model加了唯一索引，如果用戶把某條記錄刪除了，delete_status設置成1了。后來，該用戶發現不對，又重新添加了一模一樣的商品。

由于唯一索引的存在，該用戶第二次添加商品會失敗，即使該商品已經被刪除了，也沒法再添加了。

這個問題顯然有點嚴重。

有人可能會說：把name、model和delete_status三個字段同時做成唯一索引不就行了？

答：這樣做確實可以解決用戶邏輯刪除了某個商品，后來又重新添加相同的商品時，添加不了的問題。但如果第二次添加的商品，又被刪除了。該用戶第三次添加相同的商品，不也出現問題了？

由此可見，如果表中有邏輯刪除功能，是不方便創建唯一索引的。

7. 分布式鎖

接下來，你想到的第二種解決數據重復問題的辦法可能是：加分布式鎖。

目前最常用的性能最高的分布式鎖，可能是redis分布式鎖了。

使用redis分布式鎖的偽代碼如下：

try{
Stringresult=jedis.set(lockKey,requestId,"NX","PX",expireTime);
if("OK".equals(result)){
doSamething();
returntrue;
}
returnfalse;
}finally{
unlock(lockKey,requestId);
}

不過需要在finally代碼塊中釋放鎖。

其中lockKey是由商品表中的name和model組合而成的，requestId是每次請求的唯一標識，以便于它每次都能正確得釋放鎖。還需要設置一個過期時間expireTime，防止釋放鎖失敗，鎖一直存在，導致后面的請求沒法獲取鎖。

如果只是單個商品，或者少量的商品需要復制添加，則加分布式鎖沒啥問題。

主要流程如下：

可以在復制添加商品之前，先嘗試加鎖。如果加鎖成功，則在查詢商品是否存在，如果不存在，則添加商品。此外，在該流程中如果加鎖失敗，或者查詢商品時不存在，則直接返回。

加分布式鎖的目的是：保證查詢商品和添加商品的兩個操作是原子性的操作。

但現在的問題是，我們這次需要復制添加的商品數量很多，如果每添加一個商品都要加分布式鎖的話，會非常影響性能。

顯然對于批量接口，加redis分布式鎖，不是一個理想的方案。

8. 統一mq異步處理

前面我們已經聊過，在批量復制商品的接口，我們是通過RocketMQ的順序消息，單線程異步復制添加商品的，可以暫時解決商品重復的問題。

但那只改了一個添加商品的入口，還有其他添加商品的入口。

能不能把添加商品的底層邏輯統一一下，最終都調用同一段代碼。然后通過RocketMQ的順序消息，單線程異步添加商品。

主要流程如下圖所示：

這樣確實能夠解決重復商品的問題。

但同時也帶來了另外兩個問題：

現在所有的添加商品功能都改成異步了，之前同步添加商品的接口如何返回數據呢？這就需要修改前端交互，否則會影響用戶體驗。

之前不同的添加商品入口，是多線程添加商品的，現在改成只能由一個線程添加商品，這樣修改的結果導致添加商品的整體效率降低了。

由此，綜合考慮了一下各方面因素，這個方案最終被否定了。

9. insert on duplicate key update

其實，在mysql中存在這樣的語法，即：insert on duplicate key update。

在添加數據時，mysql發現數據不存在，則直接insert。如果發現數據已經存在了，則做update操作。

不過要求表中存在唯一索引或PRIMARY KEY，這樣當這兩個值相同時，才會觸發更新操作，否則是插入。

現在的問題是PRIMARY KEY是商品表的主鍵，是根據雪花算法提前生成的，不可能產生重復的數據。

但由于商品表有邏輯刪除功能，導致唯一索引在商品表中創建不了。

由此，insert on duplicate key update這套方案，暫時也沒法用。

此外，insert on duplicate key update在高并發的情況下，可能會產生死鎖問題，需要特別注意一下。

10. insert ignore

在mysql中還存在這樣的語法，即：insert ... ignore。

在insert語句執行的過程中：mysql發現如果數據重復了，就忽略，否則就會插入。

它主要是用來忽略，插入重復數據產生的Duplicate entry 'XXX' for key 'XXXX'異常的。

不過也要求表中存在唯一索引或PRIMARY KEY。

但由于商品表有邏輯刪除功能，導致唯一索引在商品表中創建不了。

由此可見，這個方案也不行。

溫馨的提醒一下，使用insert ... ignore也有可能會導致死鎖。

11. 防重表

之前聊過，因為有邏輯刪除功能，給商品表加唯一索引，行不通。

后面又說了加分布式鎖，或者通過mq單線程異步添加商品，影響創建商品的性能。

那么，如何解決問題呢？

我們能否換一種思路，加一張防重表，在防重表中增加商品表的name和model字段作為唯一索引。

例如：

CREATETABLE`product_unique`(
`id`bigint(20)NOTNULLCOMMENT'id',
`name`varchar(130)DEFAULTNULLCOMMENT'名稱',
`model`varchar(255)NOTNULLCOMMENT'規格',
`user_id`bigint(20)unsignedNOTNULLCOMMENT'創建用戶id',
`user_name`varchar(30)NOTNULLCOMMENT'創建用戶名稱',
`create_date`datetime(3)NOTNULLDEFAULTCURRENT_TIMESTAMP(3)COMMENT'創建時間',
PRIMARYKEY(`id`),
UNIQUEKEY`ux_name_model`(`name`,`model`)
)ENGINE=InnoDBDEFAULTCHARSET=utf8mb4COMMENT='商品防重表';

其中表中的id可以用商品表的id，表中的name和model就是商品表的name和model，不過在這張防重表中增加了這兩個字段的唯一索引。

視野一下子被打開了。

在添加商品數據之前，先添加防重表。如果添加成功，則說明可以正常添加商品，如果添加失敗，則說明有重復數據。

防重表添加失敗，后續的業務處理，要根據實際業務需求而定。

如果業務上允許添加一批商品時，發現有重復的，直接拋異常，則可以提示用戶：系統檢測到重復的商品，請刷新頁面重試。

例如：

try{
transactionTemplate.execute((status)->{
productUniqueMapper.batchInsert(productUniqueList);
productMapper.batchInsert(productList);
returnBoolean.TRUE;
});
}catch(DuplicateKeyExceptione){
thrownewBusinessException("系統檢測到重復的商品，請刷新頁面重試");
}

在批量插入數據時，如果出現了重復數據，捕獲DuplicateKeyException異常，轉換成BusinessException這樣運行時的業務異常。

還有一種業務場景，要求即使出現了重復的商品，也不拋異常，讓業務流程也能夠正常走下去。

例如：

try{
transactionTemplate.execute((status)->{
productUniqueMapper.insert(productUnique);
productMapper.insert(product);
returnBoolean.TRUE;
});
}catch(DuplicateKeyExceptione){
product=productMapper.query(product);
}

在插入數據時，如果出現了重復數據，則捕獲DuplicateKeyException，在catch代碼塊中再查詢一次商品數據，將數據庫已有的商品直接返回。

如果調用了同步添加商品的接口，這里非常關鍵的一點，是要返回已有數據的id，業務系統做后續操作，要拿這個id操作。

當然在執行execute之前，還是需要先查一下商品數據是否存在，如果已經存在，則直接返回已有數據，如果不存在，才執行execute方法。這一步千萬不能少。

例如：

ProductoldProduct=productMapper.query(product);
if(Objects.nonNull(oldProduct)){
returnoldProduct;
}

try{
transactionTemplate.execute((status)->{
productUniqueMapper.insert(productUnique);
productMapper.insert(product);
returnBoolean.TRUE;
});
}catch(DuplicateKeyExceptione){
product=productMapper.query(product);
}
returnproduct;

千萬注意：防重表和添加商品的操作必須要在同一個事務中，否則會出問題。

順便說一下，還需要對商品的刪除功能做特殊處理一下，在邏輯刪除商品表的同時，要物理刪除防重表。用商品表id作為查詢條件即可。

說實話，解決重復數據問題的方案挺多的，沒有最好的方案，只有最適合業務場景的，最優的方案。