RabbitMQ系列二(构建消息队列)

从AMQP协议可以看出,MessageQueue、Exchange和Binding构成了AMQP协议的核心。下面我们就围绕这三个主要组件,从应用使用的角度全面的介绍如何利用RabbitMQ构建消息队列以及使用过程中的注意事项。

声明MessageQueue:

在RabbitMQ中,无论是生产者发送消息还是消费者接收消息,都首先需要声明一个MessageQueue。这就存在一个问题,是生产者声明还是消费者声明呢?要解决这个问题,首先要明确:

1)消费者是无法订阅或者获取不存在的MessageQueue中的信息。

2)消息被Exchange接收以后,如果没有匹配的Queue,则会被丢弃。

明白了上述两点以后,就容易理解如果是消费者去声明Queue,就有可能会出现在声明Queue之前,生产者已发送的消息被丢弃的隐患。如果应用能够通过消息重发的机制允许消息丢失,则使用此方案没有任何问题。但是如果不能接受该方案,则会就需要无论是生产者还是消费者,在发送或者接收消息前,都需要去尝试建立消息队列。这里有一点需要明确,如果客户端尝试建立一个已经存在的消息队列,RabbitMQ不会做任何事情,并返回客户端建立成功的。

如果一个消费者在一个channel中正在监听某一个队列的消息,RabbitMQ是不允许该消费者在同一个channel中去声明其他队列的。RabbitMQ中,可以通过queue.declare命令声明一个队列,可以设置该队列一下属性:

1)Exclusive:排他队列,如果一个队列被声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除。这里需要注意三点:其一,排他队列是基于连接可见的,同一连接的不同信道是可以同时访问同一个连接创建的排他队列的。其二,“首次”,如果一个连接已经声明了一个排他队列,其他连接是不允许建立同名的排他队列的,这个与普通队列不同。其三,即使该队列是持久化的,一旦连接关闭或者客户端腿粗,该排他队列都会被自动删除的。这种队列适用于只限于一个客户端发送读取消息的应用场景。

2)Auto-delete:自动删除,如果该队列没有任何订阅的消费者的话,该队列会被自动删除。这种队列适用于临时队列。

3)Durable:持久化,这个会在后面作为专门讨论。

其他选项,例如如果用户仅仅想查询某一个队列是否已存在,如果不存在,不想建立该队列,仍然可以调用queue.declare,只不过需要将参数passive设为true,传给queue.declare,如果该队列已存在,则会返回true;如果不存在,则会返回Error,但是不会创建新的队列。

生产者发送消息:

在AMQP模型中,Exchange是接收生产者消息并将消息路由到消息队列的关键组件。ExchangeType和Binding决定了消息的路由规则。所以生产者想要发送消息,首先必须要声明一个Exchange和该Exchange对应的Binding。可以通过ExchangeDeclare和BindingDeclare完成。在RabbitMQ中,有三种类型的Exchange:direct、fanout和topic,不同的Exchange会表现出不同的路由行为。Durable是Exchange的持久化属性,这个会在消息持久化章节讨论。声明一个Binding需要提供一个QueueName,ExchangeName和BindingKey。

消费者订阅消息:

在RabbitMQ中消费者有2种方式获取队列中的消息:

a)一种是通过basic.consume命令,订阅某一个队列中的消息,channel会自动在处理完上一条消息之后,接收下一条消息。(同一个channel消息处理是串行的)。除非关闭channel或者取消订阅,否则客户端将会一直接收队列的消息。

b)另外一种方式是通过basic.get命令主动获取队列中的消息,但是绝对不可以通过循环调用basic.get来代替basic.consume,这是因为basic.get RabbitMQ在实际执行的时候,是首先consume某一个队列,然后检索第一条消息,然后再取消订阅。如果是高吞吐率的消费者,最好还是建议使用basic.consume。

如果有多个消费者同时订阅同一个队列的话,RabbitMQ是采用循环的方式分发消息的,每一条消息只能被一个订阅者接收。

消费者在接到消息以后,都需要给服务器发送一条确认命令。如果消费者在接到消息以后还没来得及返回ACK就断开了连接,消息服务器会重传该消息给下一个订阅者,如果没有订阅者就会存储该消息。

既然RabbitMQ提供了ACK某一个消息的命令,当然也提供了Reject某一个消息的命令。当客户端发生错误,调用basic.reject命令拒绝某一个消息时,可以设置一个requeue的属性,如果为true,则消息服务器会重传该消息给下一个订阅者;如果为false,则会直接删除该消息。当然,有可以通过ack,让消息服务器直接删除该消息并且不会重传。

持久化:

RabbitMQ默认是不持久队列、Exchange、Binding以及队列中的消息,这意味着一旦消息服务器重启,所有已声明的队列,Exchange,Binding以及队列中的消息都会丢失。通过设置Exchange和MesageQueue的durable属性为true,可以使得队列和Exchange持久化,但是这还不能使得队列中的消息持久化,这需要生产者在发送消息的时候,将delivery mode设置为2,只有这三个全部设置完成后,才能保证服务器重启不会对现有的队列造成影响。这里需要注意的是,只有durable为true的Exchange和durable为true的Queues才能绑定,否则在绑定时,RabbitMQ都会报错的。持久化对RabbitMQ的性能造成比较大的影响,可能会下降10被不止。

事务:

对事务的支持是AMQP协议的一个重要特性。假设当生产者将一个持久化消息发送给服务器时,因为consume命令本身没有任何Response返回,所以即使服务器崩溃,没有持久化该消息,生产者也无法获知该消息已经丢失。如果此时使用事务,即通过txSelect()开启一个事务,然后发送消息给服务器,然后通过txCommit()提交该事务,即可以保证,如果txCommit()提交了,则该消息一定会持久化,如果txCommit()还未提交即服务器崩溃,则该消息不会被服务器接收。当然RabbitMQ也提供了txRollback()命令用于回滚某一个事务。

Confirm机制:

使用事务固然可以保证只有提交的事务,才会被服务器执行。但是这样同时也将客户端与消息服务器同步起来,这背离了消息队列解耦的本子。RabbitMQ提供了一个更加轻量级的机制来保证生产者可以感知服务器消息是否已被路由到正确的队列中--Confirm。如果设置channel为confirm状态,则通过该channel发送的消息都会被分配一个唯一的ID,然后一旦该消息被正确的路由到匹配的队列中后,服务器会返回给生产者一个Confirm,该Confirm会包含该消息的ID,这样生产者就会知道该消息已被正确分发。对于持久化消息,只有该消息被持久化后,才会返回Confirm。Confirm机制的最大优点在于异步,生产者在发送消息以后,即可继续执行其他任务。而服务器返回Confirm后,会触发生产者的回调函数,生产者在回调函数中处理Confirm信息。如果消息服务器发生异常,导致该消息丢失,会返回给生产者一个ack,表示消息已经丢失,这样生产者就可以通过重发消息,保证消息不丢失。Confirm机制在性能上要比事务优越很多。但是Confirm机制,无法进行回滚,就是一旦服务器崩溃,生产者无法得到Confirm信息,生产者其实本身也不知道该消息已经被持久化,只有继续重发来保证消息不丢失,但是如果原先已经持久化的消息,并不会被回滚,这样队列中就会存在两条相同的消息,系统需要支持去重。

时间: 2024-10-10 12:19:22

RabbitMQ系列二(构建消息队列)的相关文章

大型网站架构系列:分布式消息队列(二)

本文是大型网站架构系列:消息队列(二),主要分享JMS消息服务,常用消息中间件(Active MQ,Rabbit MQ,Zero MQ,Kafka).[第二篇的内容大部分为网络资源的整理和汇总,供大家学习总结使用,最后有文章来源] 本次分享大纲 消息队列概述(见第一篇:大型网站架构系列:分布式消息队列(一)) 消息队列应用场景(见第一篇:大型网站架构系列:分布式消息队列(一)) 消息中间件示例(见第一篇:大型网站架构系列:分布式消息队列(一)) JMS消息服务 常用消息队列 参考(推荐)资料 本

大型网站架构系列:分布式消息队列(一) (转)

以下是消息队列以下的大纲,本文主要介绍消息队列概述,消息队列应用场景和消息中间件示例(电商,日志系统). 本次分享大纲 消息队列概述 消息队列应用场景 消息中间件示例 JMS消息服务(见第二篇:大型网站架构系列:分布式消息队列(二)) 常用消息队列(见第二篇:大型网站架构系列:分布式消息队列(二)) 参考(推荐)资料(见第二篇:大型网站架构系列:分布式消息队列(二)) 本次分享总结(见第二篇:大型网站架构系列:分布式消息队列(二)) 一.消息队列概述 消息队列中间件是分布式系统中重要的组件,主要

大型网站架构系列:分布式消息队列(一)(转)

大型网站架构系列:分布式消息队列(一) 以下是消息队列以下的大纲,本文主要介绍消息队列概述,消息队列应用场景和消息中间件示例(电商,日志系统). 本次分享大纲 消息队列概述 消息队列应用场景 消息中间件示例 JMS消息服务(见第二篇:大型网站架构系列:分布式消息队列(二)) 常用消息队列(见第二篇:大型网站架构系列:分布式消息队列(二)) 参考(推荐)资料(见第二篇:大型网站架构系列:分布式消息队列(二)) 本次分享总结(见第二篇:大型网站架构系列:分布式消息队列(二)) 一.消息队列概述 消息

大型网站架构系列:分布式消息队列(一)

以下是消息队列以下的大纲,本文主要介绍消息队列概述,消息队列应用场景和消息中间件示例(电商,日志系统). 本次分享大纲 消息队列概述 消息队列应用场景 消息中间件示例 JMS消息服务 常用消息队列 参考(推荐)资料 本次分享总结 一.消息队列概述 消息队列中间件是分布式系统中重要的组件,主要解决应用耦合,异步消息,流量削锋等问题.实现高性能,高可用,可伸缩和最终一致性架构.是大型分布式系统不可缺少的中间件. 目前在生产环境,使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ,K

消息中间件(二)消息队列

一.消息队列概述 消息队列中间件是分布式系统中重要的组件,主要解决应用耦合,异步消息,流量削锋等问题.实现高性能,高可用,可伸缩和最终一致性架构.是大型分布式系统不可缺少的中间件. 目前在生产环境,使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,RocketMQ等. 二.消息队列应用场景 以下介绍消息队列在实际应用中常用的使用场景.异步处理,应用解耦,流量削锋和消息通讯四个场景. 2.1异步处理 场景说明:用户注册后,需要发注册邮件和注册短信.传统

消息队列入门(二)消息队列的开源实现

消息队列入门(二)消息队列的开源实现 关于AMQP AMQP 是 Advanced Message Queuing Protocol,即高级消息队列协议.AMQP不是一个具体的消息队列实现,而 是一个标准化的消息中间件协议.目标是让不同语言,不同系统的应用互相通信,并提供一个简单统一的模型和编程接口.目前主流的ActiveMQ和RabbitMQ都支持AMQP协议. AMQP相关的角色和职责 Producer 消息生产者 一个给exchange发送消息的程序,发送方式大致是:它首先创建一个空消息,

在Node.js中使用RabbitMQ系列二 任务队列

在上一篇文章在Node.js中使用RabbitMQ系列一 Hello world我有使用一个任务队列,不过当时的场景是将消息发送给一个消费者,本篇文章我将讨论有多个消费者的场景. 其实,任务队列最核心解决的问题是避免立即处理那些耗时的任务,也就是避免请求-响应的这种同步模式.取而代之的是我们通过调度算法,让这些耗时的任务之后再执行,也就是采用异步的模式.我们需要将一条消息封装成一个任务,并且将它添加到任务队列里面.后台会运行多个工作进程(worker process),通过调度算法,将队列里的任

【系统架构】读《大型网站架构系列:分布式消息队列》整理

文章地址 拓展阅读: RabbitMQ + PHP (一)入门与安装 RabbitMQ + PHP (二)AMQP拓展安装 RabbitMQ + PHP (三)案例演示 1. 一些词汇和技术 1)Zookeeper注册中心 Storm集群 2. 核心思想 1)[概述] 消息队列中间件是分布式系统中重要的组件,主要解决应用耦合,异步消息,流量削锋等问题.实现高性能,高可用,可伸缩和最终一致性架构.是大型分布式系统不可缺少的中间件. 目前在生产环境,使用较多的消息队列有ActiveMQ,Rabbit

Kafka 消息队列系列之分布式消息队列Kafka

介绍 ApacheKafka®是一个分布式流媒体平台.这到底是什么意思呢?我们认为流媒体平台具有三个关键功能:它可以让你发布和订阅记录流.在这方面,它类似于消??息队列或企业消息传递系统.它允许您以容错方式存储记录流.它可以让您在发生记录时处理记录流.什么是卡夫卡好?它被用于两大类的应用程序:构建可在系统或应用程序之间可靠获取数据的实时流数据管道构建实时流应用程序,可以转换或响应数据流要了解卡夫卡如何做这些事情,让我们深入探索卡夫卡的能力.首先几个概念:Kafka作为一个或多个服务器上的集群运行