【架构】关于RabbitMQ

标签: js框架 | 发表时间:2014-08-01 09:36 | 作者:admin
出处:http://blog.haohtml.com

1      什么是RabbitMQ?

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

单向解耦
双向解耦(如:RPC)

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

rabbitmq_producer_consumer
使用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-key和binding-key的字符串切分成单词。这些单词之间用点隔开。它同样也会识别两个通配符:#匹配0个或者多个单词,*匹配一个单词。例如,binding key:*.stock.#匹配routing key:usd.stcok和eur.stock.db,但是不匹配stock.nana。

还有一些其他的交换器类型,如header、failover、system等,现在在当前的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时,如果queue是durable的,则会被清空。

 

2.5      集群(cluster):

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

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

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

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

 

3      使用

3.1      简易使用流程

rabbitmq_on_openstack

3.2      RabbitMQ在OpenStack中的使用

rabbitmq_openstack_2

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

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

93a49333-d780-335d-bdb4-0525c1b02f69

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

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

75a8cd9e-ed9e-30aa-861d-83b5f4db3d6e

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

198fa1f4-823e-364b-9cbc-1765be14f8c7

3.3      为什么要使用RabbitMQ?

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

04a63e10-0921-3819-b193-f5b55f6b21e2

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

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

转自: http://lynnkong.iteye.com/blog/1699684

相关 [架构 rabbitmq] 推荐:

【架构】关于RabbitMQ

- - 学习笔记
1      什么是RabbitMQ. RabbitMQ是实现AMQP(高级消息队列协议)的消息中间件的一种,最初起源于金融系统,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗. 消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然:. 例如一个日志系统,很容易使用RabbitMQ简化工作量,一个Consumer可以进行消息的正常处理,另一个Consumer负责对消息进行日志记录,只要在程序中指定两个Consumer所监听的queue以相同的方式绑定到同一个exchange即可,剩下的消息分发工作由RabbitMQ完成.

RabbitMQ (三) 发布/订阅

- - CSDN博客架构设计推荐文章
转发请标明出处: http://blog.csdn.net/lmj623565791/article/details/37657225. 本系列教程主要来自于官网入门教程的翻译,然后自己进行了部分的修改与实验,内容仅供参考. 上一篇博客中,我们实现了工作队列,并且我们的工作队列中的一个任务只会发给一个工作者,除非某个工作者未完成任务意外被杀死,会转发给另外的工作者,如果你还不了解: RabbitMQ (二)工作队列.

rabbitmq java client api详解

- - 五四陈科学院
以下内容由 [五四陈科学院]提供. AMQP协议是一个高级抽象层消息通信协议,RabbitMQ是AMQP协议的实现. 每个rabbitmq-server叫做一个Broker,等着tcp连接进入. 在rabbitmq-server进程内有Exchange,定义了这个消息的发送类型. Queue是进程内的逻辑队列,有多个,有名字.

RabbitMQ:镜像队列Mirrored queue

- - 飞翔的荷兰人
        在上一节 《RabbitMQ集群类型一:在单节点上构建built-in内置集群》中我们已经学习过:在集群环境中,队列只有元数据会在集群的所有节点同步,但队列中的数据只会存在于一个节点,数据没有冗余且容易丢,甚至在durable的情况下,如果所在的服务器节点宕机,就要等待节点恢复才能继续提供消息服务.

抽取rabbitmq网络层做的echo server

- 2sin18 - codedump
传说rabbitmq网络层实现的优雅高效,于是我就尝试着将其中的网络层抽取出来,模拟着做了一个echo服务器,代码放在这里.. rabbitmq的做法是内置状态机,通过切换callback的形式处理不同的业务,这样只有一个子进程处理一个链接,性能提高不少.. 测试这个echo服务器的客户端我使用的是telnet,telnet输入的数据会自动在后面加上”\r\n”发送到对端,于是代码中以这个来判断是否接收了一条消息,抽取出来回复给对端..

RabbitMQ关键性问题调研

- - Java - 编程语言 - ITeye博客
摘要:本篇是本人对RabbitMQ使用的关键性问题进行的调研,如性能上限、数据存储、集群等,.             具体的 RabbitMQ概念、使用方法、SpringAMQP配置,假设读者已有了基础. 1.1  RabbitMQ数据速率问题. 在边读边写的情况下:速率只与网络带宽正相关,网络使用率最高能达到接近100%,并且数据使用率很高(90%以上).

利用RabbitMQ实现分布式事务

- -
  实现要点:1、构建本地消息表及定时任务,确保消息可靠发送;2、RabbitMQ可靠消费;3、redis保证幂等.   两个服务:订单服务和消息服务.   使用springboot构建项目,相关代码如下. //设置消息发送确认回调,发送成功后更新消息表状态. //定时扫描记录表,将发送状态为0的消息再次发送,甚至可以记录重发次数,必要时人工干预,生产环境中需要单独部署定时任务.

[转][RabbitMQ+Python入门经典] 兔子和兔子窝

- lostsnow - heiyeluren的blog(黑夜路人的开源世界)
高级消息队列协议(AMQP1)是一个异步消息传递所使用的应用层协议规范. AMQP的原始用途只是为金融界提供一个可以彼此协作的消息协议,而现在的目标则是为通用消息队列架构提供通用构建工具. RabbitMQ作为一个工业级的消息队列中间件,基于AMQP协议的实现,由erlang语言编写. 本文讲解 RabbitMQ+Python 的使用.

使用django+celery+RabbitMQ实现异步执行

- MetUrMaker - idv2
RabbitMQ大家应该不陌生,著名的消息队列嘛. 可惜我最近才听说它的大名,了解之后不禁惊呼,世界上居然还有这种东西. 立刻觉得手里有了锤子,就看什么都是钉子了,主网站不愿意干的操作统统扔给RabbitMQ去做吧. 言归正传,先介绍一下这篇文章的应用场景吧. 我们知道大型网站的性能非常重要,然而有时不得不做一些相当耗时的操作.

Python 的服务器推送解决方案:Orbited + RabbitMQ

- 非狐外传 - python.cn(jobs, news)
最近公司要用到服务器推送技术,google了一下,nodejs固然好,但是公司的东西都是python搞的,. 所以选择了python的 Orbited +  RabbitMQ,无奈Orbited文档极其缺乏,所以要做下笔记. 以下都是在windows平台上搞的测试. RabbitMQ:先要装好Erlang,然后下RabbitMQ的win版exe文件安装,由于要用到stomp协议,.