聊聊高并發(fā)下如何防重?
最近測試給我提了一個bug,說我之前提供的一個批量復(fù)制商品的接口,產(chǎn)生了重復(fù)的商品數(shù)據(jù)。
追查原因之后發(fā)現(xiàn),這個事情沒想象中簡單,可以說一波多折。
1. 需求
產(chǎn)品有個需求:用戶選擇一些品牌,點(diǎn)擊確定按鈕之后,系統(tǒng)需要基于一份默認(rèn)品牌的商品數(shù)據(jù),復(fù)制出一批新的商品。
拿到這個需求時覺得太簡單了,三下五除二就搞定。
我提供了一個復(fù)制商品的基礎(chǔ)接口,給商城系統(tǒng)調(diào)用。
當(dāng)時的流程圖如下:
如果每次復(fù)制的商品數(shù)量不多,使用同步接口調(diào)用的方案問題也不大。
2. 性能優(yōu)化
但由于每次需要復(fù)制的商品數(shù)量比較多,可能有幾千。
如果每次都是用同步接口的方式復(fù)制商品,可能會有性能問題。
因此,后來我把復(fù)制商品的邏輯改成使用mq異步處理。
改造之后的流程圖:復(fù)制商品的結(jié)果還需要通知商城系統(tǒng):
這個方案看起來,挺不錯的。
但后來出現(xiàn)問題了。
3. 出問題了
測試給我們提了一個bug,說我之前提供的一個批量復(fù)制商品的接口,產(chǎn)生了重復(fù)的商品數(shù)據(jù)。
經(jīng)過追查之后發(fā)現(xiàn),商城系統(tǒng)為了性能考慮,也改成異步了。
他們沒有在接口中直接調(diào)用基礎(chǔ)系統(tǒng)的復(fù)制商品接口,而是在job中調(diào)用的。
站在他們的視角流程圖是這樣的:
用戶調(diào)用商城的接口,他們會往請求記錄表中寫入一條數(shù)據(jù),然后在另外一個job中,異步調(diào)用基礎(chǔ)系統(tǒng)的接口去復(fù)制商品。
但實(shí)際情況是這樣的:商城系統(tǒng)內(nèi)部出現(xiàn)了bug,在請求記錄表中,同一條請求產(chǎn)生了重復(fù)的數(shù)據(jù)。這樣導(dǎo)致的結(jié)果是,在job中調(diào)用基礎(chǔ)系統(tǒng)復(fù)制商品接口時,發(fā)送了重復(fù)的請求。
剛好基礎(chǔ)系統(tǒng)現(xiàn)在是使用RocketMQ異步處理的。由于商城的job一次會取一批數(shù)據(jù)(比如:20條記錄),在極短的時間內(nèi)(其實(shí)就是在一個for循環(huán)中)多次調(diào)用接口,可能存在相同的請求參數(shù)連續(xù)調(diào)用復(fù)制商品接口情況。于是,出現(xiàn)了并發(fā)插入重復(fù)數(shù)據(jù)的問題。
為什么會出現(xiàn)這個問題呢?
4. 多線程消費(fèi)
RocketMQ的消費(fèi)者,為了性能考慮,默認(rèn)是用多線程并發(fā)消費(fèi)的,最大支持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);
}
}
也就是說,如果在極短的時間內(nèi),連續(xù)發(fā)送重復(fù)的消息,就會被不同的線程消費(fèi)。
即使在代碼中有這樣的判斷:
Product oldProduct = query(hashCode);
if(oldProduct == null) {
productMapper.insert(product);
}
在插入數(shù)據(jù)之前,先判斷該數(shù)據(jù)是否已經(jīng)存在,只有不存在才會插入。
但由于在并發(fā)情況下,不同的線程都判斷商品數(shù)據(jù)不存在,于是同時進(jìn)行了插入操作,所以就產(chǎn)生了重復(fù)數(shù)據(jù)。
如下圖所示:
5. 順序消費(fèi)
為了解決上述并發(fā)消費(fèi)重復(fù)消息的問題,我們從兩方面著手:
- 商城系統(tǒng)修復(fù)產(chǎn)生重復(fù)記錄的bug。
- 基礎(chǔ)系統(tǒng)將消息改成單線程順序消費(fèi)。
我仔細(xì)思考了一下,如果只靠商城系統(tǒng)修復(fù)bug,以后很難避免不出現(xiàn)類似的重復(fù)商品問題,比如:如果用戶在極短的時間內(nèi)點(diǎn)擊創(chuàng)建商品按鈕多次,或者商城系統(tǒng)主動發(fā)起重試。
所以,基礎(chǔ)系統(tǒng)還需進(jìn)一步處理。
其實(shí)RocketMQ本身是支持順序消費(fèi)的,需要消息的生產(chǎn)者和消費(fèi)者一起改。
生產(chǎn)者改為:
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!");
}
});
重點(diǎn)是要調(diào)用rocketMQTemplate對象的asyncSendOrderly方法,發(fā)送順序消息。
消費(fèi)者改為:
@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);
}
}
接收消息的重點(diǎn)是RocketMQMessageListener注解中的consumeMode參數(shù),要設(shè)置成ConsumeMode.ORDERLY,這樣就能順序消費(fèi)消息了。
修改后關(guān)鍵流程圖如下:
兩邊都修改之后,復(fù)制商品這一塊就沒有再出現(xiàn)重復(fù)商品的問題了。
But,修完bug之后,我又思考了良久。
復(fù)制商品只是創(chuàng)建商品的其中一個入口,如果有其他入口,跟復(fù)制商品功能同時創(chuàng)建新商品呢?
不也會出現(xiàn)重復(fù)商品問題?
雖說,這種概率非常非常小。
但如果一旦出現(xiàn)重復(fù)商品問題,后續(xù)涉及到要合并商品的數(shù)據(jù),非常麻煩。
經(jīng)過這一次的教訓(xùn),一定要防微杜漸。
不管是用戶,還是自己的內(nèi)部系統(tǒng),從不同的入口創(chuàng)建商品,都需要解決重復(fù)商品創(chuàng)建問題。
那么,如何解決這個問題呢?
6. 唯一索引
解決重復(fù)商品數(shù)據(jù)問題,最快成本最低最有效的辦法是:給表建唯一索引。
想法是好的,但我們這邊有個規(guī)范就是:業(yè)務(wù)表必須都是邏輯刪除。
而我們都知道,要刪除表的某條記錄的話,如果用delete語句操作的話。
例如:
delete from product where id=123;
這種delete操作是物理刪除,即該記錄被刪除之后,后續(xù)通過sql語句基本查不出來。(不過通過其他技術(shù)手段可以找回,那是后話了)
還有另外一種是邏輯刪除,主要是通過update語句操作的。
例如:
update product set delete_status=1,edit_time=now(3)
where id=123;
邏輯刪除需要在表中額外增加一個刪除狀態(tài)字段,用于記錄數(shù)據(jù)是否被刪除。在所有的業(yè)務(wù)查詢的地方,都需要過濾掉已經(jīng)刪除的數(shù)據(jù)。
通過這種方式刪除數(shù)據(jù)之后,數(shù)據(jù)任然還在表中,只是從邏輯上過濾了刪除狀態(tài)的數(shù)據(jù)而已。
其實(shí)對于這種邏輯刪除的表,是沒法加唯一索引的。
為什么呢?
假設(shè)之前給商品表中的name和model加了唯一索引,如果用戶把某條記錄刪除了,delete_status設(shè)置成1了。后來,該用戶發(fā)現(xiàn)不對,又重新添加了一模一樣的商品。
由于唯一索引的存在,該用戶第二次添加商品會失敗,即使該商品已經(jīng)被刪除了,也沒法再添加了。
這個問題顯然有點(diǎn)嚴(yán)重。
有人可能會說:把name、model和delete_status三個字段同時做成唯一索引不就行了?
答:這樣做確實(shí)可以解決用戶邏輯刪除了某個商品,后來又重新添加相同的商品時,添加不了的問題。但如果第二次添加的商品,又被刪除了。該用戶第三次添加相同的商品,不也出現(xiàn)問題了?
由此可見,如果表中有邏輯刪除功能,是不方便創(chuàng)建唯一索引的。
7. 分布式鎖
接下來,你想到的第二種解決數(shù)據(jù)重復(fù)問題的辦法可能是:加分布式鎖。
目前最常用的性能最高的分布式鎖,可能是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代碼塊中釋放鎖。
其中l(wèi)ockKey是由商品表中的name和model組合而成的,requestId是每次請求的唯一標(biāo)識,以便于它每次都能正確得釋放鎖。還需要設(shè)置一個過期時間expireTime,防止釋放鎖失敗,鎖一直存在,導(dǎo)致后面的請求沒法獲取鎖。
如果只是單個商品,或者少量的商品需要復(fù)制添加,則加分布式鎖沒啥問題。
主要流程如下:
可以在復(fù)制添加商品之前,先嘗試加鎖。如果加鎖成功,則在查詢商品是否存在,如果不存在,則添加商品。此外,在該流程中如果加鎖失敗,或者查詢商品時不存在,則直接返回。
加分布式鎖的目的是:保證查詢商品和添加商品的兩個操作是原子性的操作。
但現(xiàn)在的問題是,我們這次需要復(fù)制添加的商品數(shù)量很多,如果每添加一個商品都要加分布式鎖的話,會非常影響性能。
顯然對于批量接口,加redis分布式鎖,不是一個理想的方案。
8. 統(tǒng)一mq異步處理
前面我們已經(jīng)聊過,在批量復(fù)制商品的接口,我們是通過RocketMQ的順序消息,單線程異步復(fù)制添加商品的,可以暫時解決商品重復(fù)的問題。
但那只改了一個添加商品的入口,還有其他添加商品的入口。
能不能把添加商品的底層邏輯統(tǒng)一一下,最終都調(diào)用同一段代碼。然后通過RocketMQ的順序消息,單線程異步添加商品。
主要流程如下圖所示:這樣確實(shí)能夠解決重復(fù)商品的問題。
但同時也帶來了另外兩個問題:
- 現(xiàn)在所有的添加商品功能都改成異步了,之前同步添加商品的接口如何返回數(shù)據(jù)呢?這就需要修改前端交互,否則會影響用戶體驗(yàn)。
- 之前不同的添加商品入口,是多線程添加商品的,現(xiàn)在改成只能由一個線程添加商品,這樣修改的結(jié)果導(dǎo)致添加商品的整體效率降低了。
由此,綜合考慮了一下各方面因素,這個方案最終被否定了。
9. insert on duplicate key update
其實(shí),在mysql中存在這樣的語法,即:insert on duplicate key update。
在添加數(shù)據(jù)時,mysql發(fā)現(xiàn)數(shù)據(jù)不存在,則直接insert。如果發(fā)現(xiàn)數(shù)據(jù)已經(jīng)存在了,則做update操作。
不過要求表中存在唯一索引或PRIMARY KEY,這樣當(dāng)這兩個值相同時,才會觸發(fā)更新操作,否則是插入。
現(xiàn)在的問題是PRIMARY KEY是商品表的主鍵,是根據(jù)雪花算法提前生成的,不可能產(chǎn)生重復(fù)的數(shù)據(jù)。
但由于商品表有邏輯刪除功能,導(dǎo)致唯一索引在商品表中創(chuàng)建不了。
由此,insert on duplicate key update這套方案,暫時也沒法用。
此外,insert on duplicate key update在高并發(fā)的情況下,可能會產(chǎn)生死鎖問題,需要特別注意一下。
感興趣的小伙伴,也可以找我私聊。
其實(shí)insert on duplicate key update的實(shí)戰(zhàn),我在另一篇文章《??我用kafka兩年踩過的一些非比尋常的坑??》中介紹過的,感興趣的小伙伴,可以看看。
10. insert ignore
在mysql中還存在這樣的語法,即:insert ... ignore。
在insert語句執(zhí)行的過程中:mysql發(fā)現(xiàn)如果數(shù)據(jù)重復(fù)了,就忽略,否則就會插入。
它主要是用來忽略,插入重復(fù)數(shù)據(jù)產(chǎn)生的Duplicate entry 'XXX' for key 'XXXX'異常的。
不過也要求表中存在唯一索引或PRIMARY KEY。
但由于商品表有邏輯刪除功能,導(dǎo)致唯一索引在商品表中創(chuàng)建不了。
由此可見,這個方案也不行。
溫馨的提醒一下,使用insert ... ignore也有可能會導(dǎo)致死鎖。
11. 防重表
之前聊過,因?yàn)橛羞壿媱h除功能,給商品表加唯一索引,行不通。
后面又說了加分布式鎖,或者通過mq單線程異步添加商品,影響創(chuàng)建商品的性能。
那么,如何解決問題呢?
我們能否換一種思路,加一張防重表,在防重表中增加商品表的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 '規(guī)格',
`user_id` bigint(20) unsigned NOT NULL COMMENT '創(chuàng)建用戶id',
`user_name` varchar(30) NOT NULL COMMENT '創(chuàng)建用戶名稱',
`create_date` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '創(chuàng)建時間',
PRIMARY KEY (`id`),
UNIQUE KEY `ux_name_model` (`name`,`model`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品防重表';
其中表中的id可以用商品表的id,表中的name和model就是商品表的name和model,不過在這張防重表中增加了這兩個字段的唯一索引。
視野一下子被打開了。
在添加商品數(shù)據(jù)之前,先添加防重表。如果添加成功,則說明可以正常添加商品,如果添加失敗,則說明有重復(fù)數(shù)據(jù)。
防重表添加失敗,后續(xù)的業(yè)務(wù)處理,要根據(jù)實(shí)際業(yè)務(wù)需求而定。
如果業(yè)務(wù)上允許添加一批商品時,發(fā)現(xiàn)有重復(fù)的,直接拋異常,則可以提示用戶:系統(tǒng)檢測到重復(fù)的商品,請刷新頁面重試。
例如:
try {
transactionTemplate.execute((status) -> {
productUniqueMapper.batchInsert(productUniqueList);
productMapper.batchInsert(productList);
return Boolean.TRUE;
});
} catch(DuplicateKeyException e) {
throw new BusinessException("系統(tǒng)檢測到重復(fù)的商品,請刷新頁面重試");
}
在批量插入數(shù)據(jù)時,如果出現(xiàn)了重復(fù)數(shù)據(jù),捕獲DuplicateKeyException異常,轉(zhuǎn)換成BusinessException這樣運(yùn)行時的業(yè)務(wù)異常。
還有一種業(yè)務(wù)場景,要求即使出現(xiàn)了重復(fù)的商品,也不拋異常,讓業(yè)務(wù)流程也能夠正常走下去。
例如:
try {
transactionTemplate.execute((status) -> {
productUniqueMapper.insert(productUnique);
productMapper.insert(product);
return Boolean.TRUE;
});
} catch(DuplicateKeyException e) {
product = productMapper.query(product);
}
在插入數(shù)據(jù)時,如果出現(xiàn)了重復(fù)數(shù)據(jù),則捕獲DuplicateKeyException,在catch代碼塊中再查詢一次商品數(shù)據(jù),將數(shù)據(jù)庫已有的商品直接返回。
如果調(diào)用了同步添加商品的接口,這里非常關(guān)鍵的一點(diǎn),是要返回已有數(shù)據(jù)的id,業(yè)務(wù)系統(tǒng)做后續(xù)操作,要拿這個id操作。
當(dāng)然在執(zhí)行execute之前,還是需要先查一下商品數(shù)據(jù)是否存在,如果已經(jīng)存在,則直接返回已有數(shù)據(jù),如果不存在,才執(zhí)行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;
千萬注意:防重表和添加商品的操作必須要在同一個事務(wù)中,否則會出問題。
順便說一下,還需要對商品的刪除功能做特殊處理一下,在邏輯刪除商品表的同時,要物理刪除防重表。用商品表id作為查詢條件即可。
說實(shí)話,解決重復(fù)數(shù)據(jù)問題的方案挺多的,沒有最好的方案,只有最適合業(yè)務(wù)場景的,最優(yōu)的方案。
此外,如果你對重復(fù)數(shù)據(jù)衍生出的冪等性問題感興趣的話,可以看看我的另一篇文章《??高并發(fā)下如何保證接口的冪等性???》,里面有非常詳細(xì)的介紹。