RabbitMQ消息分发轮询和Message Acknowledgment

一、消息分发

  RabbitMQ中的消息都只能存储在Queue中,生产者(下图中的P)生产消息并最终投递到Queue中,消费者(下图中的C)可以从Queue中获取消息并消费。

  

  多个消费者可以订阅同一个Queue,这时Queue中的消息会被平均分摊给多个消费者进行处理,而不是每个消费者都收到所有的消息并处理。

  

  启动3个消费者

   

  生产者依次生成3条消息

  

  

  

  可见3条消息分别被3个消费者获取,所以RabbitMQ是采用轮询机制将消息队列Queue中的消息依次发给不同的消费者

二、消息确认(Message Acknowledgment)

  在实际应用中,可能会发生消费者收到Queue中的消息,但没有处理完成就宕机(或出现其他意外)的情况,这种情况下就可能会导致消息丢失。为了避免这种情况发生,我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ,RabbitMQ收到消息回执(Message acknowledgment)后才将该消息从Queue中移除;如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开,则RabbitMQ会将该消息发送给其他消费者(如果存在多个消费者)进行处理。这里不存在timeout概念,一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者,除非它的RabbitMQ连接断开。

  如何来实现呢?只需要将consumer消费者端中 no_ack = True去掉就行了

  

  no_ack 就 no acknowlegment的意思,这个参数会导致RabbitMQ并不关心消费者有没有处理完成,可能在消费者获取消息后就将该消息从Queue中移除。去掉这个参数,如果在消费者执行过程当初出现了意外(宕机),RabbitMQ没有收到消息回执,就会发送给其他消费者执行。

 

  修改consumer端

def callback(ch, method, properties, body):
    print(‘--->>‘, ch, ‘\n‘, method, ‘\n‘, properties)
    time.sleep(30)  # 让消费者处理的时间长一点,可以用来模拟运行中断开的情况
    print(" [x] Received %r" % body)

# ch: 声明的管道channel对象内存地址
#

channel.basic_consume(callback,  # 如果收到消息就调用callback函数来处理消息
                      queue=‘hello‘,
                      # no_ack=True
                      )

   运行3个消费者,接收生产者的数据,依次关闭消费者1和消费者2,最后RabbitMQ中的消息还是会被消费者3处理。  

   

  

  

原文地址:https://www.cnblogs.com/bigberg/p/8137068.html

时间: 2024-08-30 04:52:04

RabbitMQ消息分发轮询和Message Acknowledgment的相关文章

RabbitMQ消息分发轮询

我们首先下载pika,以及rabbitMQ,和ir语言,rabbitMQ是由ir语言编写的 消息队列的使用过程大概如下: (1)客户端连接到消息队列服务器,打开一个channel. channel:消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务. (2)客户端声明一个exchange,并设置相关属性. Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列. (3)客户端声明一个queue,并设置相关属性. Queue:消息队列载体,每个

java用while循环设计轮询线程的性能问题

java用while循环设计轮询线程的性能问题 轮询线程在开发过程中的应用是比较广泛的,在这我模拟一个场景,有一个队列和轮询线程,主线程往队列中入队消息,轮询线程循环从队列中读取消息并打印消息内容.有点类似Android中Handler发送消息. 首先定义一个Message类. public class Message { private String content; public Message(String content) { this.content=content; } public

RabbitMQ学习第二记:工作队列的两种分发方式,轮询分发(Round-robin)和 公平分发(Fair dispatch)

1.什么是RabbitMQ工作队列 我们在应用程序使用消息系统时,一般情况下生产者往队列里插入数据时速度是比较快的,但是消费者消费数据往往涉及到一些业务逻辑处理导致速度跟不上生产者生产数据.因此如果一个生产者对应一个消费者的话,很容易导致很多消息堆积在队列里.这时,就得使用工作队列了.一个队列有多个消费者同时消费数据. 下图取自于官方网站(RabbitMQ)的工作队列的图例 P:消息的生产者 C1:消息的消费者1 C2:消息的消费者2 红色:队列 生产者将消息发送到队列,多个消费者同时从队列中获

(转)RabbitMQ消息队列(三):任务分发机制

在上篇文章中,我们解决了从发送端(Producer)向接收端(Consumer)发送“Hello World”的问题.在实际的应用场景中,这是远远不够的.从本篇文章开始,我们将结合更加实际的应用场景来讲解更多的高级用法. 当有Consumer需要大量的运算时,RabbitMQ Server需要一定的分发机制来balance每个Consumer的load.试想一下,对于web application来说,在一个很多的HTTP request里是没有时间来处理复杂的运算的,只能通过后台的一些工作线程

RabbitMQ消息队列(三):任务分发机制

在上篇文章中,我们解决了从发送端(Producer)向接收端(Consumer)发送“Hello World”的问题.在实际的应用场景中,这是远远不够的.从本篇文章开始,我们将结合更加实际的应用场景来讲解更多的高级用法. 当有Consumer需要大量的运算时,RabbitMQ Server需要一定的分发机制来balance每个Consumer的load.试想一下,对于web application来说,在一个很多的HTTP request里是没有时间来处理复杂的运算的,只能通过后台的一些工作线程

rabbitmq 公平分发和消息接收确认(转载)

原文地址:http://www.jianshu.com/p/f63820fe2638 当生产者投递消息到broker,rabbitmq把消息分发到消费者. 如果设置了autoAck=true 消费者会自动确认收到信息.这时broker会立即将消息删除,这种情况下如果消费者出现异常(连接中断)该消息就会丢失.为了保证消息能够被正确的消费,rabbitmq支持消息确认. String basicConsume(String queue, boolean autoAck, Consumer callb

RabbitMQ消息队列(三):任务分发机制[转]

在上篇文章中,我们解决了从发送端(Producer)向接收端(Consumer)发送“Hello World”的问题.在实际的应用场景中,这是远远不够的.从本篇文章开始,我们将结合更加实际的应用场景来讲解更多的高级用法. 当有Consumer需要大量的运算时,RabbitMQ Server需要一定的分发机制来balance每个Consumer的load.接下来我们分布讲解. 应用场景就是RabbitMQ Server会将queue的Message分发给不同的Consumer以处理计算密集型的任务

关于android 消息轮询处理

android 中涉及到服务器中数据变化信息通知用户一般有两种 办法,推送和轮询,消息推送是服务端主动发消息给客户端,因为第一时间知道数据变化是服务器自己,所以推送的优势是实时性高,但服务器主动推送需要开发一套能让客户端持久链接的服务器 现在已经有很多开源的代码实现了基于XMMP 协议的推送方案,而且还可以使用谷歌的推送方案,但有些情况并不需要服务端主动推送二是在一定的时间间隔客户端发起查询 private MyThread myThread; private NotificationManag

Android学习系列(7)--App轮询服务器消息

这篇文章是android开发人员的必备知识. 1.轮询服务器     一般的应用,定时通知消息可以采用轮询的方法从服务器拿取消息,当然实时消息通知的话,建议采用推送服务.    其中需要注意轮询的频率设置,要在需求和性能中平衡. 2.独立进程     无论程序是否正在运行,我们都要能通知到客户,我们需要一个独立进程的后台服务.     我们需要一个独立进程的后台服务.     在AndroidManifest.xml中注册Service时,有一个android:process属性,如果这个属性以