最近在做一任務(wù)時，遇到需要延遲處理的數(shù)據(jù)，最開始的做法是現(xiàn)將數(shù)據(jù)存儲在數(shù)據(jù)庫，然后寫個腳本，隔五分鐘掃描數(shù)據(jù)表再處理數(shù)據(jù)，實(shí)際效果并不好。因?yàn)橄到y(tǒng)本身一直在用RabbitMQ做異步處理任務(wù)的中間件，所以想到是否可以利用RabbitMQ實(shí)現(xiàn)延遲隊(duì)列。功夫不負(fù)有心人，RabbitMQ雖然沒有現(xiàn)成可用的延遲隊(duì)列，但是可以利用其兩個重要特性來實(shí)現(xiàn)之：1、Time To Live(TTL)消息超時機(jī)制；2、Dead Letter Exchanges（DLX）死信隊(duì)列。下面將具體描述實(shí)現(xiàn)原理以及實(shí)現(xiàn)代

延遲隊(duì)列的基礎(chǔ)原理Time To Live(TTL)

RabbitMQ可以針對Queue設(shè)置x-expires 或者 針對Message設(shè)置 x-message-ttl，來控制消息的生存時間，如果超時(兩者同時設(shè)置以最先到期的時間為準(zhǔn))，則消息變?yōu)閐ead letter(死信)RabbitMQ消息的過期時間有兩種方法設(shè)置。

通過隊(duì)列（Queue）的屬性設(shè)置，隊(duì)列中所有的消息都有相同的過期時間。（本次延遲隊(duì)列采用的方案）對消息單獨(dú)設(shè)置，每條消息TTL可以不同。

如果同時使用，則消息的過期時間以兩者之間TTL較小的那個數(shù)值為準(zhǔn)。消息在隊(duì)列的生存時間一旦超過設(shè)置的TTL值，就成為死信（dead letter）

Dead Letter Exchanges（DLX）

RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key（可選）兩個參數(shù)，如果隊(duì)列內(nèi)出現(xiàn)了dead letter，則按照這兩個參數(shù)重新路由轉(zhuǎn)發(fā)到指定的隊(duì)列。

x-dead-letter-exchange：出現(xiàn)死信（dead letter）之后將dead letter重新發(fā)送到指定exchange x-dead-letter-routing-key：出現(xiàn)死信（dead letter）之后將dead letter重新按照指定的routing-key發(fā)送

隊(duì)列中出現(xiàn)死信（dead letter）的情況有：

消息或者隊(duì)列的TTL過期。（延遲隊(duì)列利用的特性） 隊(duì)列達(dá)到最大長度 消息被消費(fèi)端拒絕（basic.reject or basic.nack）并且requeue=false

綜合上面兩個特性，將隊(duì)列設(shè)置TTL規(guī)則，隊(duì)列TTL過期后消息會變成死信，然后利用DLX特性將其轉(zhuǎn)發(fā)到另外的交換機(jī)和隊(duì)列就可以被重新消費(fèi)，達(dá)到延遲消費(fèi)效果。

延遲隊(duì)列設(shè)計(jì)及實(shí)現(xiàn)（Python）

從上面描述，延遲隊(duì)列的實(shí)現(xiàn)大致分為兩步：

產(chǎn)生死信，有兩種方式Per-Message TTL和 Queue TTL，因?yàn)槲业男枨笾惺撬械南⒀舆t處理時間相同，所以本實(shí)現(xiàn)中采用 Queue TTL設(shè)置隊(duì)列的TTL，如果需要將隊(duì)列中的消息設(shè)置不同的延遲處理時間，則設(shè)置Per-Message TTL（官方文檔）

設(shè)置死信的轉(zhuǎn)發(fā)規(guī)則,Dead Letter Exchanges設(shè)置方法（官方文檔）

完整代碼如下：

'''Created on Fri Aug 3 17:00:44 2018@author: Bge'''import pika,json,loggingclass RabbitMQClient: def __init__(self, conn_str=’amqp://user:pwd@host:port/%2F’): self.exchange_type = 'direct' self.connection_string = conn_str self.connection = pika.BlockingConnection(pika.URLParameters(self.connection_string)) self.channel = self.connection.channel() self._declare_retry_queue() #RetryQueue and RetryExchange logging.debug('connection established') def close_connection(self): self.connection.close() logging.debug('connection closed') def declare_exchange(self, exchange): self.channel.exchange_declare(exchange=exchange, exchange_type=self.exchange_type, durable=True) def declare_queue(self, queue): self.channel.queue_declare(queue=queue, durable=True,) def declare_delay_queue(self, queue,DLX=’RetryExchange’,TTL=60000): ''' 創(chuàng)建延遲隊(duì)列 :param TTL: ttl的單位是us，ttl=60000 表示 60s :param queue: :param DLX:死信轉(zhuǎn)發(fā)的exchange :return: ''' arguments={} if DLX: #設(shè)置死信轉(zhuǎn)發(fā)的exchange arguments[ ’x-dead-letter-exchange’]=DLX if TTL: arguments[’x-message-ttl’]=TTL print(arguments) self.channel.queue_declare(queue=queue, durable=True, arguments=arguments) def _declare_retry_queue(self): ''' 創(chuàng)建異常交換器和隊(duì)列，用于存放沒有正常處理的消息。 :return: ''' self.channel.exchange_declare(exchange=’RetryExchange’, exchange_type=’fanout’, durable=True) self.channel.queue_declare(queue=’RetryQueue’, durable=True) self.channel.queue_bind(’RetryQueue’, ’RetryExchange’,’RetryQueue’) def publish_message(self,routing_key, msg,exchange=’’,delay=0,TTL=None): ''' 發(fā)送消息到指定的交換器 :param exchange: RabbitMQ交換器 :param msg: 消息實(shí)體，是一個序列化的JSON字符串 :return: ''' if delay==0: self.declare_queue(routing_key) else: self.declare_delay_queue(routing_key,TTL=TTL) if exchange!=’’: self.declare_exchange(exchange) self.channel.basic_publish(exchange=exchange, routing_key=routing_key, body=msg, properties=pika.BasicProperties( delivery_mode=2, type=exchange )) self.close_connection() print('message send out to %s' % exchange) logging.debug('message send out to %s' % exchange) def start_consume(self,callback,queue=’#’,delay=1): ''' 啟動消費(fèi)者，開始消費(fèi)RabbitMQ中的消息 :return: ''' if delay==1: queue=’RetryQueue’ else: self.declare_queue(queue) self.channel.basic_qos(prefetch_count=1) try: self.channel.basic_consume( # 消費(fèi)消息callback, # 如果收到消息，就調(diào)用callback函數(shù)來處理消息queue=queue, # 你要從那個隊(duì)列里收消息 ) self.channel.start_consuming() except KeyboardInterrupt: self.stop_consuming() def stop_consuming(self): self.channel.stop_consuming() self.close_connection() def message_handle_successfully(channel, method): ''' 如果消息處理正常完成，必須調(diào)用此方法， 否則RabbitMQ會認(rèn)為消息處理不成功，重新將消息放回待執(zhí)行隊(duì)列中 :param channel: 回調(diào)函數(shù)的channel參數(shù) :param method: 回調(diào)函數(shù)的method參數(shù) :return: ''' channel.basic_ack(delivery_tag=method.delivery_tag) def message_handle_failed(channel, method): ''' 如果消息處理失敗，應(yīng)該調(diào)用此方法，會自動將消息放入異常隊(duì)列 :param channel: 回調(diào)函數(shù)的channel參數(shù) :param method: 回調(diào)函數(shù)的method參數(shù) :return: ''' channel.basic_reject(delivery_tag=method.delivery_tag, requeue=False)

發(fā)布消息代碼如下：

from MQ.RabbitMQ import RabbitMQClientprint('start program')client = RabbitMQClient()msg1 = ’{'key':'value'}’client.publish_message(’test-delay’,msg1,delay=1,TTL=10000)print('message send out')

消費(fèi)者代碼如下：

from MQ.RabbitMQ import RabbitMQClientimport jsonprint('start program')client = RabbitMQClient()def callback(ch, method, properties, body): msg = body.decode() print(msg) # 如果處理成功，則調(diào)用此消息回復(fù)ack，表示消息成功處理完成。 RabbitMQClient.message_handle_successfully(ch, method)queue_name = 'RetryQueue'client.start_consume(callback,queue_name,delay=0)

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。