【workbunny/webman-rabbitmq】消費(fèi)者超時(shí)導(dǎo)致重復(fù)消費(fèi)

Caesar-Tang 更新于2023-05-17

問(wèn)題描述

首先感謝 workbunny 提供這么一款很好用的插件。使用【workbunny】RabbitMQ客戶端，插件地址：http://m.wtbis.cn/plugin/67, 在消費(fèi)者中如果有阻塞并超過(guò)一定時(shí)間，就會(huì)導(dǎo)致重復(fù)消費(fèi)。

框架及插件版本

"workerman/webman-framework": "^1.5.0"
"workbunny/webman-rabbitmq": "^1.0"

復(fù)現(xiàn)代碼

生產(chǎn)者：IndexController.php

workbunny rabbitmq配置：app.php

消費(fèi)者：TestBuilder.php

關(guān)鍵日志打印

關(guān)鍵日志說(shuō)明

通過(guò)日志可以看到，在出現(xiàn)錯(cuò)誤前，是有消息消費(fèi)成功并返回了 ACK。在出現(xiàn)錯(cuò)誤后，相關(guān)進(jìn)程重啟，導(dǎo)致消息開(kāi)始重新消費(fèi)，以前返回的 ACK 似乎無(wú)效？RabbitMQ的控制臺(tái)，始終顯示有對(duì)應(yīng)數(shù)量的信息處于 Unacked 狀態(tài)。

 2430  3 2

3個(gè)回答

tanhongbin 2023-05-18

消費(fèi)也要做冪等性

Caesar-Tang 2023-05-18

使用冪等的確能彌補(bǔ)重復(fù)消費(fèi)，但需要引入新的服務(wù)比如數(shù)據(jù)庫(kù)或緩存；測(cè)試發(fā)現(xiàn)一條消息會(huì)存在不止重復(fù)消費(fèi)一次的情況，即使使用冪等但由此帶來(lái)的阻塞和性能消耗也是未知的。

chaz6chez 2023-05-18

workbunny/webman-rabbitmq是基于bunny的一個(gè)異步rabbitmq-client，底層依賴event-loop，相當(dāng)于把所有事件都投送到loop中執(zhí)行，你在消費(fèi)函數(shù)里使用sleep會(huì)出讓當(dāng)前進(jìn)程，讓當(dāng)前進(jìn)程的event-loop暫時(shí)停轉(zhuǎn)等待cpu調(diào)度；
您把sleep換成

$time = time();
while(time() <= $time + 1) {}

來(lái)模擬阻塞試試是否還會(huì)出現(xiàn)重復(fù)消費(fèi)的情況；

另外在實(shí)際使用上，消費(fèi)者邏輯建議增加冪等性的判斷。

最后感謝您的建議和意見(jiàn)，我后續(xù)會(huì)根據(jù)目前的猜測(cè)論證一下，然后優(yōu)化這個(gè)問(wèn)題；
最近也是比較忙，2.0.0-beta也擱置了一段時(shí)間了

chaz6chez 2023-05-18

所有的行為都是異步的，包括ack，在打印了ack后，程序?qū)ck行為投放到event-loop，然后進(jìn)程殺死，event-loop中的ack消息可能還沒(méi)有真正的執(zhí)行，但event-loop已經(jīng)隨當(dāng)前進(jìn)程殺死了
chaz6chez 2023-05-18

這個(gè)也是目前2.X開(kāi)發(fā)的一個(gè)方向
Caesar-Tang 2023-05-18
添加冪等性判斷后，會(huì)存在這樣一種情況，僅供參考，測(cè)試用例描述以下：

生產(chǎn)者邏輯：發(fā)送 N 條消息

消費(fèi)者邏輯：首先從緩存中讀取該條消息的消費(fèi)記錄，若有，直接返回 ACK，沒(méi)有，進(jìn)入消費(fèi)處理（耗時(shí)操作）后返回 ACK

現(xiàn)象：消費(fèi)第一條消息，處理完耗時(shí)操作后返回 ACK，異常，進(jìn)程重啟，然后重復(fù)處理第 1 條，因?yàn)榫彺嬷幸呀?jīng)有記錄過(guò)消費(fèi)，直接返回 ACK；開(kāi)始處理第 2 條... ；到最后 1 條拋出異常后，進(jìn)程重啟，會(huì)發(fā)現(xiàn)前面幾條已經(jīng)直接返回 ACK 的又會(huì)被重復(fù)消費(fèi)了。

以前在項(xiàng)目中也使用過(guò)該插件，十分好用。最近在業(yè)務(wù)上做批量發(fā)送消息（發(fā)送消息是阻塞的）發(fā)現(xiàn)了該問(wèn)題。也了解到作者技術(shù)強(qiáng)并且樂(lè)于分享。在此表示感謝。
chaz6chez 2023-05-18

這個(gè)也是目前2.x版本開(kāi)發(fā)要解決的問(wèn)題之一，因?yàn)樗行袨槎际钱惒降囊院髸?huì)帶來(lái)很多行為會(huì)滯后執(zhí)行，那么就沒(méi)辦法把控，導(dǎo)致一些意外情況的產(chǎn)生；比如異步publish后進(jìn)程被殺死，可能導(dǎo)致數(shù)據(jù)實(shí)際上沒(méi)有投放成功；消費(fèi)者成功消費(fèi)了以后返回ack，但進(jìn)程因意外被殺死，可能導(dǎo)致消息重復(fù)消費(fèi)。
因?yàn)槟壳?.x屬于能將就用就將就用狀態(tài)，現(xiàn)目前是全力開(kāi)發(fā)2.x以解決之前已知的這些問(wèn)題，所以你可以描述一下的的業(yè)務(wù)需求，我看看嘗試給你出個(gè)方案，可能不一定使用rabbitmq。
Caesar-Tang 2023-05-19
當(dāng)前的業(yè)務(wù)場(chǎng)景，由管理端給微信用戶推送訂閱信息：

一次推送會(huì)有萬(wàn)以上的數(shù)量

微信官方不支持批量推送訂閱消息，只能由應(yīng)用端一個(gè)個(gè)給用戶下發(fā)，即請(qǐng)求微信接口

記錄每次的推送結(jié)果（可以分批）

嘗試的方案（在選擇方案前還未發(fā)現(xiàn)超時(shí)重復(fù)消費(fèi)問(wèn)題）：

使用消息中間件 rabbitmq

由管理端發(fā)布推送任務(wù)，根據(jù)本次推送任務(wù)獲取推送用戶，將用戶分批后，按批發(fā)布待消費(fèi)消息并在數(shù)據(jù)庫(kù)記錄本次消息發(fā)布結(jié)果（生產(chǎn)者）

消費(fèi)者消費(fèi)，循環(huán)給該批用戶發(fā)送微信訂閱消息，結(jié)束后更新數(shù)據(jù)庫(kù)消費(fèi)結(jié)果（消費(fèi)者）

為了避免超時(shí)，推送微信訂閱消息時(shí)設(shè)置接口請(qǐng)求超時(shí)時(shí)間，合理控制每批的用戶數(shù)量。但由此每批用戶數(shù)量只能比較少。
chaz6chez 2023-05-19
根據(jù)你們的業(yè)務(wù)來(lái)看，貌似沒(méi)有需要頻繁關(guān)閉/重啟消費(fèi)進(jìn)程的地方，如果是進(jìn)程持續(xù)運(yùn)行的話，實(shí)際上不存在重復(fù)消費(fèi)的問(wèn)題，重復(fù)消費(fèi)的問(wèn)題只存在于關(guān)閉進(jìn)程的那一瞬間；

方案一：盡可能減少重復(fù)消費(fèi)的方案：

將QOS設(shè)置為1，即每個(gè)消費(fèi)進(jìn)程消費(fèi)1條數(shù)據(jù)；

消費(fèi)進(jìn)程開(kāi)啟cpu count * 2；

盡可能避免關(guān)閉消費(fèi)者進(jìn)程或者重啟消費(fèi)者進(jìn)程的操作

方案二：換其他隊(duì)列組件試試
可以先暫時(shí)測(cè)試一下http://m.wtbis.cn/plugin/69是否滿足你們的需求，這個(gè)隊(duì)列同樣具備ack機(jī)制，同時(shí)也經(jīng)歷過(guò)小范圍生產(chǎn)測(cè)試，內(nèi)部的執(zhí)行方式是同步執(zhí)行