文档章节

什么是RabbitMQ

 如风达
发布于 2015/01/12 18:11
字数 2408
阅读 80
收藏 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
博文 256
码字总数 23157
作品 0
深圳
私信 提问
RabbitMQ 入门指南——安装

RabbitMQ好文 Rabbitmq Java Client Api详解 tohxyblog-博客园-rabbitMQ教程系列 robertohuang-CSDN-rabbitMQ教程系列 RabbitMQ 中文文档 RabbitMQ 能为你做些什么? RabbitMQ Quick(快速手...

osc_gtphseny
2019/03/03
4
0
RabbitMQ简单介绍+Windows环境安装

文章目录 1、RabbitMQ简介2、RabbitMQ与其他MQ有什么不同3、RabbitMQ环境安装3.1 安装erlang3.2 安装rabbitmq-server4. RabbitMQ管理平台介绍 1、RabbitMQ简介 RabbitMQ 是一个由 erlang 开发...

osc_veu13fnv
2019/08/30
3
0
RabbitMQ TroubleShooting

RabbitMQ是一款优秀的消息队列中间件,提供了稳定、监控完善的产品,但是软件就会有bug。为了前进路径可以畅通,我们必须了解出现的一些故障的快速处理方式,毕竟在生产环境,时间就是生命,...

osc_ymlf86ez
2018/10/28
8
0
Windos下安装RabbitMQ

RabbitMQ是什么 ? RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统。他遵循Mozilla Public License开源协议。 1.安装RabbitMQ需要先安装Erlang语言开发包。(注意Erlang的版本必须...

osc_bkhfgue7
01/14
1
0
在C#中使用消息队列RabbitMQ

1、什么是RabbitMQ。详见 http://www.rabbitmq.com/。 作用就是提高系统的并发性,将一些不需要及时响应客户端且占用较多资源的操作,放入队列,再由另外一个线程,去异步处理这些队列,可极...

osc_veu13fnv
2019/05/11
1
0

没有更多内容

加载失败,请刷新页面

加载更多

华为pat地址转换,以及内网web服务器发布

实验名称:华为pat地址转换,以及内网web服务器发布 实验拓扑图: 3. 实验目的 : 1.使内网通过pat转化出去上外网 2.使用静态pat做端口映射,发布web服务器 3.配置交换机远程登录 4. 地址规划 cl...

osc_2ch77h9m
1分钟前
0
0
php优化与模块

缓存 xcache eaccelerator zendopcache pdo_mysql扩展 memcache扩展 memcached扩展 mongo扩展 imagemagick imagick扩展...

osc_tb68dlqx
2分钟前
0
0
ATFX投资者教育活动凸显优势,助力客户有效应对市场波动

ATFX动态:新冠肺炎疫情全球传播期间,各国纷纷颁布“居家令”等措施,要求员工必须待在家里,因此也给各行各业带来了不同程度的影响。为了配合各国颁布的防疫措施,ATFX全球办事处依据各地的...

ATFX
3分钟前
10
0
ospf的工作原理

#谈起ospf(开放式最短路径优先),首先它是一种链路状态路由协议 它的工作过程主要有以下几点 : # 建立邻居表 hello : 仅仅使用hello报文,发现邻居,用于邻居的建立,维护,拆除 ospf 路由仅仅...

osc_4dgu16li
4分钟前
16
0
mysql5.7.20安装

rpm -qa | grep mysql yum -y install make gcc-c++ cmake bison-devel ncurses-devel cd /root wget http://www.sourceforge.net/projects/boost/files/boost/1.59.0/boost_1_59_0.tar.gz m......

osc_pl4ni83h
5分钟前
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部