最近測試給我提了一個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
public class MessageReceiver implements RocketMQListener<MessageExt> {
@Override
public void onMessage(MessageExt message) {
String message = new String(message.getBody(), StandardCharsets.UTF_8);
doSamething(message);
}
}
也就是說，如果在極短的時間内，連續發送重複的消息，就會被不同的線程消費。
即使在代碼中有這樣的判斷：
Product oldProduct = query(hashCode);
if(oldProduct == null) {
productMapper.insert(product);
}
在插入數據之前，先判斷該數據是否已經存在，隻有不存在才會插入。
但由于在并發情況下，不同的線程都判斷商品數據不存在，于是同時進行了插入操作，所以就産生了重複數據。
如下圖所示：
為了解決上述并發消費重複消息的問題，我們從兩方面着手：
我仔細思考了一下，如果隻靠商城系統修複bug，以後很難避免不出現類似的重複商品問題，比如：如果用戶在極短的時間内點擊創建商品按鈕多次，或者商城系統主動發起重試。
所以，基礎系統還需進一步處理。
其實RocketMQ本身是支持順序消費的，需要消息的生産者和消費者一起改。
生産者改為：
rocketMQTemplate.asyncSendOrderly(topic, message, hashKey, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
log.info("sendMessage success");
}
@Override
public void onException(Throwable e) {
log.error("sendMessage failed!");
}
});
重點是要調用rocketMQTemplate對象的asyncSendOrderly方法，發送順序消息。
消費者改為：
@RocketMQMessageListener(topic = "${com.susan.topic:PRODUCT_TOPIC}",
consumeMode = ConsumeMode.ORDERLY,
consumerGroup = "${com.susan.group:PRODUCT_TOPIC_GROUP}")
@Service
public class MessageReceiver implements RocketMQListener<MessageExt> {
@Override
public void onMessage(MessageExt message) {
String message = new String(message.getBody(), StandardCharsets.UTF_8);
doSamething(message);
}
}
接收消息的重點是RocketMQMessageListener注解中的consumeMode參數，要設置成ConsumeMode.ORDERLY，這樣就能順序消費消息了。
修改後關鍵流程圖如下：
兩邊都修改之後，複制商品這一塊就沒有再出現重複商品的問題了。
But，修完bug之後，我又思考了良久。
複制商品隻是創建商品的其中一個入口，如果有其他入口，跟複制商品功能同時創建新商品呢？
不也會出現重複商品問題？
雖說，這種概率非常非常小。
但如果一旦出現重複商品問題，後續涉及到要合并商品的數據，非常麻煩。
經過這一次的教訓，一定要防微杜漸。
不管是用戶，還是自己的内部系統，從不同的入口創建商品，都需要解決重複商品創建問題。
那麼，如何解決這個問題呢？
6. 唯一索引解決重複商品數據問題，最快成本最低最有效的辦法是：給表建唯一索引。
想法是好的，但我們這邊有個規範就是：業務表必須都是邏輯删除。
而我們都知道，要删除表的某條記錄的話，如果用delete語句操作的話。
例如：
delete from product where id=123;
這種delete操作是物理删除，即該記錄被删除之後，後續通過sql語句基本查不出來。（不過通過其他技術手段可以找回，那是後話了）
還有另外一種是邏輯删除，主要是通過update語句操作的。
例如：
update product set delete_status=1,edit_time=now(3)
where id=123;
邏輯删除需要在表中額外增加一個删除狀态字段，用于記錄數據是否被删除。在所有的業務查詢的地方，都需要過濾掉已經删除的數據。
通過這種方式删除數據之後，數據任然還在表中，隻是從邏輯上過濾了删除狀态的數據而已。
其實對于這種邏輯删除的表，是沒法加唯一索引的。
為什麼呢？
假設之前給商品表中的name和model加了唯一索引，如果用戶把某條記錄删除了，delete_status設置成1了。後來，該用戶發現不對，又重新添加了一模一樣的商品。
由于唯一索引的存在，該用戶第二次添加商品會失敗，即使該商品已經被删除了，也沒法再添加了。
這個問題顯然有點嚴重。
有人可能會說：把name、model和delete_status三個字段同時做成唯一索引不就行了？
答：這樣做确實可以解決用戶邏輯删除了某個商品，後來又重新添加相同的商品時，添加不了的問題。但如果第二次添加的商品，又被删除了。該用戶第三次添加相同的商品，不也出現問題了？
由此可見，如果表中有邏輯删除功能，是不方便創建唯一索引的。
5. 分布式鎖接下來，你想到的第二種解決數據重複問題的辦法可能是：加分布式鎖。
目前最常用的性能最高的分布式鎖，可能是redis分布式鎖了。
使用redis分布式鎖的僞代碼如下：
try{
String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
if ("OK".equals(result)) {
doSamething();
return true;
}
return false;
} finally {
unlock(lockKey,requestId);
}
不過需要在finally代碼塊中釋放鎖。
其中lockKey是由商品表中的name和model組合而成的，requestId是每次請求的唯一标識，以便于它每次都能正确得釋放鎖。還需要設置一個過期時間expireTime，防止釋放鎖失敗，鎖一直存在，導緻後面的請求沒法獲取鎖。
如果隻是單個商品，或者少量的商品需要複制添加，則加分布式鎖沒啥問題。
主要流程如下：
可以在複制添加商品之前，先嘗試加鎖。如果加鎖成功，則在查詢商品是否存在，如果不存在，則添加商品。此外，在該流程中如果加鎖失敗，或者查詢商品時不存在，則直接返回。
加分布式鎖的目的是：保證查詢商品和添加商品的兩個操作是原子性的操作。
但現在的問題是，我們這次需要複制添加的商品數量很多，如果每添加一個商品都要加分布式鎖的話，會非常影響性能。
顯然對于批量接口，加redis分布式鎖，不是一個理想的方案。
6. 統一mq異步處理前面我們已經聊過，在批量複制商品的接口，我們是通過RocketMQ的順序消息，單線程異步複制添加商品的，可以暫時解決商品重複的問題。
但那隻改了一個添加商品的入口，還有其他添加商品的入口。
能不能把添加商品的底層邏輯統一一下，最終都調用同一段代碼。然後通過RocketMQ的順序消息，單線程異步添加商品。
主要流程如下圖所示：
這樣确實能夠解決重複商品的問題。
但同時也帶來了另外兩個問題：
由此，綜合考慮了一下各方面因素，這個方案最終被否定了。
7. 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在高并發的情況下，可能會産生死鎖問題，需要特别注意一下。
感興趣的小夥伴，也可以找我私聊。
其實insert on duplicate key update的實戰，我在另一篇文章《我用kafka兩年踩過的一些非比尋常的坑》中介紹過的，感興趣的小夥伴，可以看看。
8. insert ignore在mysql中還存在這樣的語法，即：insert ... ignore。
在insert語句執行的過程中：mysql發現如果數據重複了，就忽略，否則就會插入。
它主要是用來忽略，插入重複數據産生的Duplicate entry 'XXX' for key 'XXXX'異常的。
不過也要求表中存在唯一索引或PRIMARY KEY。
但由于商品表有邏輯删除功能，導緻唯一索引在商品表中創建不了。
由此可見，這個方案也不行。
溫馨的提醒一下，使用insert ... ignore也有可能會導緻死鎖。
9. 防重表之前聊過，因為有邏輯删除功能，給商品表加唯一索引，行不通。
後面又說了加分布式鎖，或者通過mq單線程異步添加商品，影響創建商品的性能。
那麼，如何解決問題呢？
我們能否換一種思路，加一張防重表，在防重表中增加商品表的name和model字段作為唯一索引。
例如：
CREATE TABLE `product_unique` (
`id` bigint(20) NOT NULL COMMENT 'id',
`name` varchar(130) DEFAULT NULL COMMENT '名稱',
`model` varchar(255) NOT NULL COMMENT '規格',
`user_id` bigint(20) unsigned NOT NULL COMMENT '創建用戶id',
`user_name` varchar(30) NOT NULL COMMENT '創建用戶名稱',
`create_date` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '創建時間',
PRIMARY KEY (`id`),
UNIQUE KEY `ux_name_model` (`name`,`model`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品防重表';
其中表中的id可以用商品表的id，表中的name和model就是商品表的name和model，不過在這張防重表中增加了這兩個字段的唯一索引。
視野一下子被打開了。
在添加商品數據之前，先添加防重表。如果添加成功，則說明可以正常添加商品，如果添加失敗，則說明有重複數據。
防重表添加失敗，後續的業務處理，要根據實際業務需求而定。
如果業務上允許添加一批商品時，發現有重複的，直接抛異常，則可以提示用戶：系統檢測到重複的商品，請刷新頁面重試。
例如：
try {
transactionTemplate.execute((status) -> {
productUniqueMapper.batchInsert(productUniqueList);
productMapper.batchInsert(productList);
return Boolean.TRUE;
});
} catch(DuplicateKeyException e) {
throw new BusinessException("系統檢測到重複的商品，請刷新頁面重試");
}
在批量插入數據時，如果出現了重複數據，捕獲DuplicateKeyException異常，轉換成BusinessException這樣運行時的業務異常。
還有一種業務場景，要求即使出現了重複的商品，也不抛異常，讓業務流程也能夠正常走下去。
例如：
try {
transactionTemplate.execute((status) -> {
productUniqueMapper.insert(productUnique);
productMapper.insert(product);
return Boolean.TRUE;
});
} catch(DuplicateKeyException e) {
product = productMapper.query(product);
}
在插入數據時，如果出現了重複數據，則捕獲DuplicateKeyException，在catch代碼塊中再查詢一次商品數據，将數據庫已有的商品直接返回。
如果調用了同步添加商品的接口，這裡非常關鍵的一點，是要返回已有數據的id，業務系統做後續操作，要拿這個id操作。
當然在執行execute之前，還是需要先查一下商品數據是否存在，如果已經存在，則直接返回已有數據，如果不存在，才執行execute方法。這一步千萬不能少。
例如：
Product oldProduct = productMapper.query(product);
if(Objects.nonNull(oldProduct)) {
return oldProduct;
}
try {
transactionTemplate.execute((status) -> {
productUniqueMapper.insert(productUnique);
productMapper.insert(product);
return Boolean.TRUE;
});
} catch(DuplicateKeyException e) {
product = productMapper.query(product);
}
return product;
千萬注意：防重表和添加商品的操作必須要在同一個事務中，否則會出問題。
順便說一下，還需要對商品的删除功能做特殊處理一下，在邏輯删除商品表的同時，要物理删除防重表。用商品表id作為查詢條件即可。
說實話，解決重複數據問題的方案挺多的，沒有最好的方案，隻有最适合業務場景的，最優的方案。
此外，如果你對重複數據衍生出的幂等性問題感興趣的話，可以看看我的另一篇文章《高并發下如何保證接口的幂等性？》，裡面有非常詳細的介紹。
,
更多精彩资讯请关注tft每日頭條，我们将持续为您更新最新资讯!
zpostcode 
Recruit 
weather 
mreligion 
Yellowpages 
sport 
constellation 
shopping 
name 
game 
directory 
literature 
Word 
tour 
furnish 
Lottery 
tftnews 
lyrics 
News 
digital 
car 
dir 
Edu 
Finance 
