文档章节

什么是RabbitMQ

 如风达
发布于 2015/01/12 18:11
字数 2408
阅读 27
收藏 0

1      什么是RabbitMQ

RabbitMQ是实现AMQP(高级消息队列协议)的消息中间件的一种,最初起源于金融系统,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然:


 

单向解耦


 

双向解耦(如:RPC

    例如一个日志系统,很容易使用RabbitMQ简化工作量,一个Consumer可以进行消息的正常处理,另一个Consumer负责对消息进行日志记录,只要在程序中指定两个Consumer所监听的queue以相同的方式绑定到同一个exchange即可,剩下的消息分发工作由RabbitMQ完成。
 

 

使用RabbitMQ server需要:

1. ErLang语言包;

2. RabbitMQ安装包;

RabbitMQ同时提供了java的客户端(一个jar包)。

 

2      概念和特性

2.1      交换机(exchange):

1. 接收消息,转发消息到绑定的队列。四种类型:direct, topic, headers and fanout

direct:转发消息到routigKey指定的队列

topic:按规则转发消息(最灵活)

headers:(这个还没有接触到)

fanout:转发消息到所有绑定队列

2. 如果没有队列绑定在交换机上,则发送到该交换机上的消息会丢失。

3. 一个交换机可以绑定多个队列,一个队列可以被多个交换机绑定。

4. topic类型交换器通过模式匹配分析消息的routing-key属性。它将routing-keybinding-key的字符串切分成单词。这些单词之间用点隔开。它同样也会识别两个通配符:#匹配0个或者多个单词,*匹配一个单词。例如,binding key*.stock.#匹配routing keyusd.stcokeur.stock.db,但是不匹配stock.nana

还有一些其他的交换器类型,如headerfailoversystem等,现在在当前的RabbitMQ版本中均未实现。

5. 因为交换器是命名实体,声明一个已经存在的交换器,但是试图赋予不同类型是会导致错误。客户端需要删除这个已经存在的交换器,然后重新声明并且赋予新的类型。

6. 交换器的属性:

持久性:如果启用,交换器将会在server重启前都有效。

自动删除:如果启用,那么交换器将会在其绑定的队列都被删除掉之后自动删除掉自身。

惰性:如果没有声明交换器,那么在执行到使用的时候会导致异常,并不会主动声明。

 

2.2      队列(queue):

1. 队列是RabbitMQ内部对象,存储消息。相同属性的queue可以重复定义。

2. 临时队列。channel.queueDeclare(),有时不需要指定队列的名字,并希望断开连接时删除队列。

3. 队列的属性:

持久性:如果启用,队列将会在server重启前都有效。

自动删除:如果启用,那么队列将会在所有的消费者停止使用之后自动删除掉自身。

惰性:如果没有声明队列,那么在执行到使用的时候会导致异常,并不会主动声明。

排他性:如果启用,队列只能被声明它的消费者使用。

这些性质可以用来创建例如排他和自删除的transient或者私有队列。这种队列将会在所有链接到它的客户端断开连接之后被自动删除掉。它们只是短暂地连接到server,但是可以用于实现例如RPC或者在AMQ上的对等通信。4. RPC的使用是这样的:RPC客户端声明一个回复队列,唯一命名(例如用UUID),并且是自删除和排他的。然后它发送请求给一些交换器,在消息的reply-to字段中包含了之前声明的回复队列的名字。RPC服务器将会回答这些请求,使用消息的reply-to作为routing key(默认绑定器会绑定所有的队列到默认交换器,名称为“amp.交换器类型名”)发送到默认交换器。注意这仅仅是惯例而已,可以根据和RPC服务器的约定,它可以解释消息的任何属性(甚至数据体)来决定回复给谁。

2.3      消息传递:

1. 消息在队列中保存,以轮询的方式将消息发送给监听消息队列的消费者,可以动态的增加消费者以提高消息的处理能力。

2. 为了实现负载均衡,可以在消费者端通知RabbitMQ,一个消息处理完之后才会接受下一个消息。

channel.basic_qos(prefetch_count=1)

注意:要防止如果所有的消费者都在处理中,则队列中的消息会累积的情况。

3. 消息有14个属性,最常用的几种:

deliveryMode:持久化属性

contentType:编码

replyTo:指定一个回调队列

correlationId:消息id

实例代码:

4. 消息生产者可以选择是否在消息被发送到交换器并且还未投递到队列(没有绑定器存在)和/或没有消费者能够立即处理的时候得到通知。通过设置消息的mandatory/immediate属性为真,这些投递保障机制的能力得到了强化。

5. 此外,一个生产者可以设置消息的persistent属性为真。这样一来,server将会尝试将这些消息存储在一个稳定的位置,直到server崩溃。当然,这些消息肯定不会被投递到非持久的队列中。

 

2.4      高可用性(HA):

1. 消息ACK,通知RabbitMQ消息已被处理,可以从内存删除。如果消费者因宕机或链接失败等原因没有发送ACK(不同于ActiveMQ,在RabbitMQ里,消息没有过期的概念),则RabbitMQ会将消息重新发送给其他监听在队列的下一个消费者。

channel.basicConsume(queuename, noAck=false, consumer);

2. 消息和队列的持久化。定义队列时可以指定队列的持久化属性(问:持久化队列如何删除?)

channel.queueDeclare(queuename, durable=true, false, false, null);

发送消息时可以指定消息持久化属性:

channel.basicPublish(exchangeName, routingKey,

            MessageProperties.PERSISTENT_TEXT_PLAIN,

            message.getBytes());

这样,即使RabbitMQ服务器重启,也不会丢失队列和消息。

3. publisher confirms

4. master/slave机制,配合Mirrored Queue,这种情况下,publisher会正常发送消息和接收消息的confirm,但对于subscriber来说,需要接收Consumer Cancellation Notifications来得到主节点失败的通知,然后re-consume from the queue,此时要求client有处理重复消息的能力。注意:如果queue在一个新加入的节点上增加了一个slave,此时slave上没有此前queue的信息(目前还没有同步机制)。

(通过命令行或管理插件可以查看哪个slave是同步的:

rabbitmqctl list_queues name slave_pids synchronised_slave_pids

    当一个slave重新加入mirrored-queue时,如果queuedurable的,则会被清空。

 

2.5      集群(cluster):

1. 不支持跨网段(如需支持,需要shovelfederation插件)

2. 可以随意的动态增加或减少、启动或停止节点,允许节点故障

3. 集群分为RAM节点和DISK节点,一个集群最好至少有一个DISK节点保存集群的状态。

4. 集群的配置可以通过命令行,也可以通过配置文件,命令行优先。

 

3      使用

3.1      简易使用流程
 

3.2      RabbitMQOpenStack中的使用

 

 

    Openstack中,组件之间对RabbitMQ使用基本都是“Remote Procedure Calls”的方式。每一个Nova服务(比如计算服务、存储服务等)初始化时会创建两个队列,一个名为“NODE-TYPE.NODE-ID”,另一个名为“NODE-TYPE”,NODE-TYPE是指服务的类型,NODE-ID指节点名称。

    从抽象层面上讲,RabbitMQ的组件的使用类似于下图所示:

每个服务会绑定两个队列到同一个topic类型的exchange,从不同的队列中接收不同类型的消息。消息的发送者如果关心消息的返回值,则会监听另一个队列,该队列绑定在一个direct类型的exchange。接受者收到消息并处理后,会将消息的返回发送到此exchange

Openstack中,如果不关心消息返回,消息的流程图如下:

 

    如果关心消息返回值,流程图如下:



 

 

3.3      为什么要使用RabbitMQ

曾经有过一个人做过一个测试(http://www.cnblogs.com/amityat/archive/2011/08/31/2160293.html),发送1百万个并发消息,对性能有很高的需求,于是作者对比了RabbitMQMSMQActiveMQZeroMQueue,整个过程共产生1百万条1K的消息。测试的执行是在一个Windows Vista上进行的,测试结果如下:



    虽然ZeroMQ性能较高,但这个产品不提供消息持久化,需要自己实现审计和数据恢复,因此在易用性和HA上不是令人满意,通过测试结果可以看到,RabbitMQ的性能确实不错。

    我在本机也做了一些测试,但我的测试是基于组件的原生配置,没有做任何的配置优化,因此总觉的不靠谱。我只测试了RabbitMQActiveMQ两款产品,虽然网上都说ActiveMQ性能不如前者,但平心而论,ActiveMQ提供了很多配置,存在很大的调优空间,也许修改一个配置参数就会使组件的性能有一个质的飞跃。


本文转载自:

共有 人打赏支持
粉丝 7
博文 255
码字总数 22313
作品 0
深圳
在CentOS上安装rabbitmq

转自:http://flyingdutchman.iteye.com/blog/1887283 这文章写得很好,除了安装软件编译时间比较长之外,安装这个几乎没出现什么错误。现在去配置下rabbitmq,马上就可以使用了。 在本节中我...

mac_zhao
2014/09/28
0
0
【原创】rabbitmq-echopid用户手册(翻译)

为了方便工作中使用,周末抽空对 rabbitmq-echopid 用户手册进行了翻译,鉴于自己水平有限,翻译中难免有纰漏产生,如果疑问,欢迎指出探讨。 官方原文:http://www.rabbitmq.com/man/rabbi...

摩云飞
2012/11/12
0
0
RabbitMQ详解(二)------消息通信的概念

  PS:近期在南宁出差,工作比较忙,所以更新会比较慢。   说到消息通信,可能我们首先会想到的是邮箱,QQ,微信,短信等等这些通信方式,这些通信方式都有发送者,接收者,还有一个中间...

ysocean
07/09
0
0
RabbitMQ 3.7.5-beta.3 发布,带来多处 bug 修复

RabbitMQ 3.7.5-beta.3 发布,此版本是维护版本的预览版,主要带来了多处 bug 修复,涉及模块包括: Core Server CLI Tools Management Plugin LDAP Plugin Shovel Plugin Peer Discovery A...

雨田桑
04/30
0
0
RabbitMQ集群整合SpringBoot2.x

RabbitMQ相信大家已经再熟悉不过了,作为业界四大主流消息中间件之一(Apache RocketMQ、Apache Kafka、Apache ActiveMQ、RabbitMQ),它具有非常好的性能和可靠性的集群模式,不仅仅在各大互...

阿神_
07/24
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Kafka如何做到1秒处理1500万条消息?

Apache Kafka 是一款流行的分布式数据流平台,它已经广泛地被诸如 New Relic(数据智能平台)、Uber、Square(移动支付公司)等大型公司用来构建可扩展的、高吞吐量的、且高可靠的实时数据流系统...

架构师springboot
4分钟前
0
0
如何清理Nexus Repository Manager

随着开发运维一体化的理念逐渐深入,持续集成流水线已经成为软件开发的标配。持续集成通过高度自动化的方式,使得构建发布变得非常简单,构建发布次数随之变的越来越多。我们在享受高频次的构...

JasonSE
17分钟前
0
0
聊聊storm的ack机制

序 本文主要研究一下storm的ack机制 实例 SentenceSpout public class AckSentenceSpout extends BaseRichSpout { private ConcurrentHashMap<UUID, Values> pending; private Spo......

go4it
26分钟前
0
0
《netty入门与实战》笔记-05:netty内置的channelHandler

Netty 内置了很多开箱即用的 ChannelHandler。下面,我们通过学习 Netty 内置的 ChannelHandler 来逐步构建我们的 pipeline。 ChannelInboundHandlerAdapter 与 ChannelOutboundHandlerAdap...

Funcy1122
46分钟前
3
0
帧动画(wifi信号动态动画)

准备六张wifi不同信号强度的素材图片,复制到drawable目录中 在drawable目录中新建frame文件,并编写代码 <animation-list xmlns:android="http://schemas.android.com/apk/res/android"> ...

lanyu96
今天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部