哈尔滨网站制作开发报价,搭建网上购物商城的步骤,中车建设工程有限公司网站,聊城做网站信息最近在做一任务时#xff0c;遇到需要延迟处理的数据#xff0c;最开始的做法是现将数据存储在数据库#xff0c;然后写个脚本#xff0c;隔五分钟扫描数据表再处理数据#xff0c;实际效果并不好。因为系统本身一直在用rabbitmq做异步处理任务的中间件#xff0c;所以想…最近在做一任务时遇到需要延迟处理的数据最开始的做法是现将数据存储在数据库然后写个脚本隔五分钟扫描数据表再处理数据实际效果并不好。因为系统本身一直在用rabbitmq做异步处理任务的中间件所以想到是否可以利用rabbitmq实现延迟队列。功夫不负有心人rabbitmq虽然没有现成可用的延迟队列但是可以利用其两个重要特性来实现之1、time to live(ttl)消息超时机制2、dead letter exchanges(dlx)死信队列。下面将具体描述实现原理以及实现代延迟队列的基础原理time to live(ttl)rabbitmq可以针对queue设置x-expires 或者 针对message设置 x-message-ttl来控制消息的生存时间如果超时(两者同时设置以最先到期的时间为准)则消息变为dead letter(死信)rabbitmq消息的过期时间有两种方法设置。通过队列(queue)的属性设置队列中所有的消息都有相同的过期时间。(本次延迟队列采用的方案)对消息单独设置每条消息ttl可以不同。如果同时使用则消息的过期时间以两者之间ttl较小的那个数值为准。消息在队列的生存时间一旦超过设置的ttl值就成为死信(dead letter)dead letter exchanges(dlx)rabbitmq的queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key(可选)两个参数如果队列内出现了dead letter则按照这两个参数重新路由转发到指定的队列。x-dead-letter-exchange出现死信(dead letter)之后将dead letter重新发送到指定exchangex-dead-letter-routing-key出现死信(dead letter)之后将dead letter重新按照指定的routing-key发送队列中出现死信(dead letter)的情况有消息或者队列的ttl过期。(延迟队列利用的特性)队列达到最大长度消息被消费端拒绝(basic.reject or basic.nack)并且requeuefalse综合上面两个特性将队列设置ttl规则队列ttl过期后消息会变成死信然后利用dlx特性将其转发到另外的交换机和队列就可以被重新消费达到延迟消费效果。延迟队列设计及实现(python)从上面描述延迟队列的实现大致分为两步产生死信有两种方式per-message ttl和 queue ttl因为我的需求中是所有的消息延迟处理时间相同所以本实现中采用 queue ttl设置队列的ttl如果需要将队列中的消息设置不同的延迟处理时间则设置per-message ttl()设置死信的转发规则,dead letter exchanges设置方法()完整代码如下created on fri aug 3 17:00:44 2018author: bgeimport pika,json,loggingclass rabbitmqclient:def __init__(self, conn_stramqp://user:pwdhost:port/%2f):self.exchange_type directself.connection_string conn_strself.connection pika.blockingconnection(pika.urlparameters(self.connection_string))self.channel self.connection.channel()self._declare_retry_queue() #retryqueue and retryexchangelogging.debug(connection established)def close_connection(self):self.connection.close()logging.debug(connection closed)def declare_exchange(self, exchange):self.channel.exchange_declare(exchangeexchange,exchange_typeself.exchange_type,durabletrue)def declare_queue(self, queue):self.channel.queue_declare(queuequeue,durabletrue,)def declare_delay_queue(self, queue,dlxretryexchange,ttl60000):创建延迟队列:param ttl: ttl的单位是usttl60000 表示 60s:param queue::param dlx:死信转发的exchange:return:arguments{}if dlx:#设置死信转发的exchangearguments[ x-dead-letter-exchange]dlxif ttl:arguments[x-message-ttl]ttlprint(arguments)self.channel.queue_declare(queuequeue,durabletrue,argumentsarguments)def _declare_retry_queue(self):创建异常交换器和队列用于存放没有正常处理的消息。:return:self.channel.exchange_declare(exchangeretryexchange,exchange_typefanout,durabletrue)self.channel.queue_declare(queueretryqueue,durabletrue)self.channel.queue_bind(retryqueue, retryexchange,retryqueue)def publish_message(self,routing_key, msg,exchange,delay0,ttlnone):发送消息到指定的交换器:param exchange: rabbitmq交换器:param msg: 消息实体是一个序列化的json字符串:return:if delay0:self.declare_queue(routing_key)else:self.declare_delay_queue(routing_key,ttlttl)if exchange!:self.declare_exchange(exchange)self.channel.basic_publish(exchangeexchange,routing_keyrouting_key,bodymsg,propertiespika.basicproperties(delivery_mode2,typeexchange))self.close_connection()print(message send out to %s % exchange)logging.debug(message send out to %s % exchange)def start_consume(self,callback,queue#,delay1):启动消费者开始消费rabbitmq中的消息:return:if delay1:queueretryqueueelse:self.declare_queue(queue)self.channel.basic_qos(prefetch_count1)try:self.channel.basic_consume( # 消费消息callback, # 如果收到消息就调用callback函数来处理消息queuequeue, # 你要从那个队列里收消息)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):如果消息处理正常完成必须调用此方法否则rabbitmq会认为消息处理不成功重新将消息放回待执行队列中:param channel: 回调函数的channel参数:param method: 回调函数的method参数:return:channel.basic_ack(delivery_tagmethod.delivery_tag)def message_handle_failed(channel, method):如果消息处理失败应该调用此方法会自动将消息放入异常队列:param channel: 回调函数的channel参数:param method: 回调函数的method参数:return:channel.basic_reject(delivery_tagmethod.delivery_tag, requeuefalse)发布消息代码如下from mq.rabbitmq import rabbitmqclientprint(start program)client rabbitmqclient()msg1 {key:value}client.publish_message(test-delay,msg1,delay1,ttl10000)print(message send out)消费者代码如下from mq.rabbitmq import rabbitmqclientimport jsonprint(start program)client rabbitmqclient()def callback(ch, method, properties, body):msg body.decode()print(msg)# 如果处理成功则调用此消息回复ack表示消息成功处理完成。rabbitmqclient.message_handle_successfully(ch, method)queue_name retryqueueclient.start_consume(callback,queue_name,delay0)以上就是本文的全部内容希望对大家的学习有所帮助也希望大家多多支持萬仟网。希望与广大网友互动点此进行留言吧
