建站
咨詢
售后
建站咨詢
新聞中心
紅色的消息隊(duì)列——重新尋回失去的機(jī)會(huì)
創(chuàng)新互聯(lián)主營(yíng)霍城網(wǎng)站建設(shè)的網(wǎng)絡(luò)公司,主營(yíng)網(wǎng)站建設(shè)方案,APP應(yīng)用開發(fā),霍城h5重慶小程序開發(fā)搭建,霍城網(wǎng)站營(yíng)銷推廣歡迎霍城等地區(qū)企業(yè)咨詢
隨著互聯(lián)網(wǎng)的快速發(fā)展，消息隊(duì)列作為一種重要的網(wǎng)絡(luò)通信手段，已經(jīng)成為了現(xiàn)今很多應(yīng)用系統(tǒng)中不可或缺的一部分。然而，在消息隊(duì)列中，由于網(wǎng)絡(luò)不穩(wěn)定等原因，有時(shí)也可能會(huì)出現(xiàn)消息的丟失，這種情況給應(yīng)用系統(tǒng)帶來的后果十分嚴(yán)重。我們需要一種能夠自我檢測(cè)、自我修復(fù)的消息隊(duì)列來減少這種“失靈”的概率，那么，紅色的消息隊(duì)列——可靠隊(duì)列就應(yīng)運(yùn)而生了。
先來看一下在傳統(tǒng)消息隊(duì)列中，當(dāng)我們啟動(dòng)消費(fèi)者客戶端進(jìn)行消費(fèi)時(shí)，遇到處理異?；蛘叻?wù)宕機(jī)等情況，就有可能丟失部分消息，這種丟失不僅會(huì)對(duì)應(yīng)用系統(tǒng)造成巨大的危害，更是無法檢測(cè)和追溯，一旦發(fā)生就是不可挽回的損失。
可靠隊(duì)列的出現(xiàn)，解決了這一難題。在可靠隊(duì)列中，消息的消費(fèi)不再是一次性的，而是分為三個(gè)階段，分別是：預(yù)處理（PREPARE）、提交（COMMIT）和回滾（ROLLBACK），其中預(yù)處理和提交都保證消息的可靠性，而回滾則是提供了一個(gè)修復(fù)機(jī)制。
下面，我們來簡(jiǎn)單介紹一下可靠隊(duì)列的核心原理：
1、消息預(yù)處理
在預(yù)處理階段，消費(fèi)者會(huì)首先請(qǐng)求消息隊(duì)列中的一批消息（一般是10條以上），并將這些消息標(biāo)記為“已占用”。同時(shí)，應(yīng)用系統(tǒng)會(huì)將這些消息的 ID 號(hào)批量保存到可靠隊(duì)列上，這里我們可以使用 Redis 來實(shí)現(xiàn) ID 存儲(chǔ)和數(shù)據(jù)查詢的功能。預(yù)處理階段結(jié)束后，消費(fèi)者進(jìn)行批量處理，將處理成功的消息標(biāo)記為已處理，而處理失敗的消息則會(huì)被重寫入隊(duì)列中。
2、消息提交
在消息處理成功后，消費(fèi)者將進(jìn)行消息的提交，并移除進(jìn)入預(yù)處理隊(duì)列中的消息 ID。如果消費(fèi)者無法成功提交消息，則這些消息的狀態(tài)依舊是“已占用”，后續(xù)其他消費(fèi)者可以看作“消息未完成”，進(jìn)行重復(fù)消費(fèi)，直到成功提交為止。
3、消息回滾
在消息消費(fèi)者的處理失敗或者異常退出時(shí)，這些消息仍舊會(huì)被留在可靠隊(duì)列中，而不會(huì)被消除。然后，系統(tǒng)會(huì)自動(dòng)將這些消息狀態(tài)進(jìn)行回滾，并放回消息隊(duì)列中供其他消費(fèi)者消費(fèi)。
除了以上三種核心原理之外，可靠隊(duì)列還有其他比較重要的功能，如：消息去重、消息重復(fù)消費(fèi)、消息延時(shí)消費(fèi)等，這些功能的實(shí)現(xiàn)也是提高消息隊(duì)列可靠性的重要手段。
下面，我們就來實(shí)際編寫可靠隊(duì)列中幾個(gè)重要的功能點(diǎn)：
1、消息去重
在消費(fèi)者端進(jìn)行去重是非常必要的，因?yàn)樵诓l(fā)情況下，可能會(huì)有多個(gè)消費(fèi)者同時(shí)在處理同一消息，如果不做去重處理，則會(huì)造成消息的重復(fù)消費(fèi)，增大系統(tǒng)負(fù)擔(dān)，因此，我們需要利用 Redis 數(shù)據(jù)庫進(jìn)行消息 ID 去重的處理，請(qǐng)參考以下代碼片段：
“`python
import redis
class ReliableQueue(object):
def __init__(self):
self.rd = redis.Redis(host=’localhost’, port=6379)
#去重
def remove_duplication(self, message_id):
return self.rd.sadd(‘message_id_set’, message_id)
2、消息重復(fù)消費(fèi)
在可靠隊(duì)列中，消息的消費(fèi)者不只一個(gè)，那么當(dāng)一些消息處理失敗或者出現(xiàn)異常的時(shí)候，需要其他消費(fèi)者對(duì)這些故障消息進(jìn)行重復(fù)消費(fèi)，我們也需要在消費(fèi)者端進(jìn)行對(duì)應(yīng)的設(shè)置，請(qǐng)參考以下代碼片段：
```python
import time
import redis
class ReliableQueue(object):
def __init__(self):
self.rd = redis.Redis(host='localhost', port=6379)
#消息重復(fù)消費(fèi)
def repeat_message(self, message_id, retry_time):
count = 0
while count
if self.rd.sismember('message_id_set', message_id):
self.rd.srem('message_id_set', message_id)
break
else:
count += 1
time.sleep(1)
3、消息延時(shí)消費(fèi)
在實(shí)際業(yè)務(wù)場(chǎng)景中，可能會(huì)有某些任務(wù)需要延時(shí)執(zhí)行，以達(dá)到更好的效果。針對(duì)這種需求，我們可以在消費(fèi)者端設(shè)置帶有過期時(shí)間的 Redis 鍵，具體代碼實(shí)現(xiàn)如下：
“`python
import redis
class ReliableQueue(object):
def __init__(self):
self.rd = redis.Redis(host=’localhost’, port=6379)
#消息延時(shí)
def message_delay(self, msg, delay_time):
self.rd.set(msg[‘message_id’], msg)
self.rd.expire(msg[‘message_id’], delay_time)
綜上所述，可靠隊(duì)列作為一種高可用、高并發(fā)的消息中間件，為應(yīng)用系統(tǒng)提供了可靠的消息傳輸機(jī)制。在不斷發(fā)展的過程中，我們也需要不斷優(yōu)化和完善它的功能，以應(yīng)對(duì)日益復(fù)雜的應(yīng)用場(chǎng)景。
創(chuàng)新互聯(lián)服務(wù)器托管擁有成都T3+級(jí)標(biāo)準(zhǔn)機(jī)房資源，具備完善的安防設(shè)施、三線及BGP網(wǎng)絡(luò)接入帶寬達(dá)10T，機(jī)柜接入千兆交換機(jī)，能夠有效保證服務(wù)器托管業(yè)務(wù)安全、可靠、穩(wěn)定、高效運(yùn)行；創(chuàng)新互聯(lián)專注于成都服務(wù)器托管租用十余年，得到成都等地區(qū)行業(yè)客戶的一致認(rèn)可。
新聞名稱：紅色的消息隊(duì)列重新尋回失去的機(jī)會(huì)（redis消息隊(duì)列撤銷）
網(wǎng)頁鏈接：http://www.5511xx.com/article/djpsiii.html
