消息队列及rabbitmq简介

2017-05-23 本文已影响692人 zhuke

1.什么是消息队列

消息队列允许应用间通过消息的发送与接收的方式进行通信，当消息接收方服务忙或不可用时，其提供了一个消息暂存的功能。

消息队列中的几个基本概念：

Producer: 消息发送方，也叫做消息生产者

Consumer: 消息接收方，也叫做消息消费者

Broker: 消息投递的代理

2.应用场景

低耦合、可靠投递、广播、流量控制、最终一致性。

1）解耦（为面向服务的架构（SOA）提供基本的最终一致性实现）

基于消息的模型，关心的是“通知”，而非“处理”。

短信、邮件通知、缓存刷新等操作使用消息队列进行通知。

系统性能评价标准：

非消息系统队列系统：系统中最慢的组件运行时间（短板效应）

消息队列系统：异步通知减少短板效应的影响

消息队列和RPC的区别与比较：

RPC: 异步调用，及时获得调用结果，具有强一致性结果，关心业务调用处理结果。

消息队列：两次异步RPC调用，将调用内容在队列中进行转储，并选择合适的时机进行投递（错峰流控）

2）广播（发布/订阅模型，基于消息格式进行编程）

如果没有消息队列，每当一个新的业务方接入，我们都要联调一次新接口。有了消息队列，我们只需要关心消息是否送达了队列，至于谁希望订阅，是下游的事情，无疑极大地减少了开发和联调的工作量。

3）错峰及流控

不同服务间的处理能力不同。比如：

WEB前端上千万次/s的负载，数据库tps跟不上。

没有消息队列时，通过滑动窗口、拥塞控制等一系列措施，可以做到服务调用间的流量控制，但是不具有通用性，且维护复杂。

通过控制消息队列的分发速度可以实现通用的流量控制。利用中间系统转储两个系统的通信内容，并在下游系统有能力处理这些消息的时候，再处理这些消息，是一套相对较通用的方式。

总结：

1. 消息队列适用于处理非强一致性事务要求，非延迟敏感的业务场景。RPC可满足强一致性和延迟敏感要求；

2.可以用支持最终一致性的消息队列来实现轻量级、非延迟敏感的分布式事务；

3.上下游业务系统处理能力存在较大差距，且不包含在业务主流程的功能，可以交给消息队列；

4.有多个下游业务关心系统发出的通知消息时，基于标准的消息内容格式，使用消息队列可减少重复开发和联调。

3. 消息队列的基本特性

数据流转过程：producer发送给broker, broker发送给consumer, consumer回复消费确认，broker删除消息

1）消息丢失（可靠性）

会导致消息重复及延迟。

当有可能发生消息丢失风险的操作前，先将消息数据落地，然后发送。失败或超时时，不断轮询所有待发送消息重新发送，保证最终一定送达。消息消费者收到消息后给服务端一个确认，当所有订阅者都确认收到消息后，删除消息。

消息重复和丢失必须得面对一个，只能在两者间做平衡。

2）顺序投递

1.允许消息丢失。

2.从发送方到服务方到接受者都是单点单线程。

3）重复投递

如何鉴别重复消息？

重复消息如何幂等处理？

i) 版本号

在一个回话周期内维护一个单调递增的消息版本号，检测到小于最近接收到的版本号时判别为重复投递的消息。

发送方必须携带消息版本号；

对于严格要求消息顺序的业务，接收方必须存储消息版本号。

ii）状态机

主流消息队列的设计范式里，在不丢消息的前提下，尽量减少重复消息，不保证消息的投递顺序。

4）push or pull

push 模式, 即 broker 主动推送消息给消费者

pull 模式, 即消费者主动从 broker 中拉取消息

4. rabbitmq（以下内容均基于AMQP协议）

1）基本特性

* 可靠性

消息持久化、消息发送和投递确认机制、集群高可用方案

* 灵活路由

消息通过exchange的方式路由到不同的queue中，提供了包括fanout, direct, topic等多种exchange实现，并且支持通过编写exchange插件的方式自实现路由方案

* 支持集群

同网段下的rabbitmq节点可以通过集群的方式，组成一个逻辑上的单一broker

* Federation

通过Federation可以在跨网段节点间组件集群

* 高可用消息队列

通过设置镜像队列的方式，消息可以在镜像队列间进行复制，使节点down机或硬件损坏的情况下保证队列服务的高可用

* 多协议支持

包括AMQP, STOMP, MQTT, HTTP等多种消息交换协议

* 多客户端支持

JAVA, .NET, Ruby, Python, PHP, Node, Go......

* 可视化管理界面

* 丰富的插件支持

tracing, managment-plugin, and you can also write your own.

2）基础模型

基础组件：Publisher, Exchange, Queue, Consumer

基础动作：Publish, Binding, Route, Consume

i) Queue的基本属性：

Name

Durable (the queue will survive a broker restart)

Exclusive (used by only one connection and the queue will be deleted when that connection closes)

Auto-delete (queue is deleted when last consumer unsubscribes)

Arguments (some brokers use it to implement additional features like message TTL)

ii) Exchange路由策略

direct(default):

路由到完全匹配的routing_key对应绑定的queue中

Fanout:

将消息路由转发到所有绑定到该exchange上的queue

Topic:

通过比对routing_key是否匹配来路由消息。

*(star) can substitute for exactly one word.

#(hash) can substitute for zero or more words.

Header:

忽略routingKey，通过匹配header中的键值对方式来路由，有all(全部匹配)和any(部分匹配)两种模式。

iii) Binding

绑定动作是将发送到exchange上的消息路由到特定queue的规则的绑定。

3）消息投递的可靠性保证

Consumet Acknowledgements & Publisher Confirms

i) （Consumer） Delivery Acknowledgements

当broker投递一条消息给consumer后，broker需要获取客户端是否正确处理了该条消息的状态，当所有消息订阅者都成功确认了该条消息时，证明该条消息投递成功，broker会从队列里面删除该条消息。

Delivery Identifiers: Delivery Tags

当一个consumer向服务器注册了一个连接（建立起了一个channel），broker向consumer投递消息时会携带一个作用域为channel，单调递增，long类型的delivery tag，consumer通过收到的delivery tag作为标识，向broker发起消息确认流程。

（注：delivery tag是一个64位的long类型数，其最大值为：9223372036854775807，channel作用域内超过其最大值的可能性很小）

Acknowledgements Models

basic.ack消息被成功处理

basic.nack客户端无法处理该消息，支持一次拒绝多条消息（multiple表示reject

basic.reject客户端无法处理该消息

Channel Prefetch Setting (QoS)

Consumer可以同时接收多少条消息。

ex: tag=5,6,7,8都处于未确认状态，QoS=4,那么broker将不会再向该Consumer投递任何消息，直到有一条消息被确认后。

ii) Publisher Confirms

保证消息被准确送达到broker的两种方式：

transaction & Publisher Acknowledgements

transaction

当channel处于事务模式下时，publisher投递消息至broker，broker的消息落地在同一个事务中，保证了消息从publisher到broker的确定性。

事务模式相较于普通异步模式，性能上有250倍的下降。

Publisher Acknowledgements

通过delivery tag来作为消息的唯一标识，来对publisher的消息做异步确认，对于basic.reject & nack的情况需要业务方自己对消息做暂存重试。

官方示例：http://hg.rabbitmq.com/rabbitmq-java-client/file/default/test/src/com/rabbitmq/examples/ConfirmDontLoseMessages.java

消息在什么时候被确认

消息没有找到对应的路由策略，broker会返回basic.nack表示broker不能正确处理该条消息；

对于持久化队列的消息，当消息被成功写入磁盘后，会返回basic.ack的确认消息，表示消息broker正常收到消息且数据已落地；

对于镜像队列，当所有镜像结点的对应queue已经成功接收到数据时，会返回broker.ack表示数据已成功同步；

rabbitmq的异步publisher confirm策略会将一段时间间隔内（a few hundred milliseconds）的所有message写入到磁盘，所有异步confirm大概会有几百毫秒的延迟，但是这样却带来了吞吐量的极大提高。

4）消息投递的顺序

当消息从单publisher产生，通过单个exchange路由到单个queue，并被单个consumer消费时保证消息是严格按照顺序消费的。

当消息被多个consumer订阅时，c1消费成功，c2消费失败且进行了requeue操作，此时就打乱了原来的顺序。

5）TTL(time to live)

消息存活时间，当一个消息从入队开始计时，过了设定的TTL时间后，就认为该消息为dead message.

6）DLX(Dead Letter Exchange)

死信队列。当发生如下情况时，消息会进入DLX:

The message is rejected (basic.rejectorbasic.nack) withrequeue=false,

The TTL for the message expires; or

The queue length limit is exceeded.

总结

rabbitmq保证如下特性：

消息不丢失

消息至少被成功消费一次

不保证消息不重复，需业务方自行进行去重逻辑（版本号&状态机&msgid）

不保证完全按照顺序消费

5）高可用方案

cluster + ha policy

cluster机制

多个全联通节点之间元信息(exchange、queue、binding等)保持强一致，但是队列消息只会存储在其中一个节点。

优点：提高吞吐量，部分解决扩展性问题。

缺点：不能提升数据可靠性和系统可用性。

ha policy机制

在cluster机制基础上可以指定集群内任意数量队列组成镜像队列，队列消息会在多节点间复制。实现数据高可靠和系统高可用。

设置参数：ha-mode和ha-params可以细粒度(哪些节点，哪些队列)设置镜像队列。

设置参数：ha-sync-mode=manual(默认)/automatic可以指定集群中新节点的数据同步策略。

home node:rabbitmq中每一个queue都有一个home node，称之为queue master，所有对queue的操作都是首先通过queue master向其他节点进行复制的，该机制保证了队列的FIFO特性。

我们可以通过设置queue的x-queue-master-locator设置queue master策略

Pick the node hosting the minimum number of masters:min-masters

Pick the node the client that declares the queue is connected to:client-local

Pick a random node:random

消息节点故障的几种情况

slave节点故障

slave节点故障时，集群会自动关将其剔除，不再将master的信息同步至该节点。

new node joining the cluster.

节点可在任意时间加入集群，初始化加入集群的slave队列信息为空，但是可以接收从master新同步过来的消息。master队列头部消息不断被消费，尾部不断新增消息，当某一时刻，在slave加入之前的历史消息都被消费完毕时，master队列的size和slave的size相等，认为slave节点从master的数据同步完成。

新加入的slave不能为加入之前的数据增加额外的冗余和可用性。因为执行明确的同步操作会使queue无响应，因此只允许非活跃的queues进行明确的同步操作而活跃的queues进行自然的同步操作是一种好的策略。

也可以在新的slave加入集群后，手动或自动触发同步历史数据，但是在同步过程中，队列queue将处于不可用状态（类似于MySQL在更改表结构的时候的锁表）。

rabbitmqctl sync_queuename 手动触发历史消息同步

rabbitmqctl cancel_sync_queuename 取消消息同步操作

rabbitmqctl list_queues name slave_pids synchronised_slave_pids 查看正在同步的队列

通过设置队列的ha-sync-mode属性automatic-新slave加入自动同步，manual-通过增量的方式同步

Stopping nodes and synchronisation

master节点故障，最先加入集群的slave将被提升为master，假设此时集群中存在一个已经完全同步的节点，即最先加入集群的slave。

集群中的节点继续故障，到最终只剩下唯一的一个节点，该节点为master，对于该节点上持久化队列，会在其故障或者重启前不断持久化消息，当故障节点恢复后，重新加入节点，因为其无法得知其故障之前持久化的镜像队列的历史消息是否被正确投递，rabbitmq采取的策略是清空其持久化信息，作为一个空的新slave加入集群。

Stopping master nodes with only unsynchronised slaves

当master节点故障时，cluster中只有未同步完成的节点时，有以下两种情况：

当在可控条件下故障：rabbitmqctl stop, 优雅地关闭OS时，故障节点将不会转移到镜像队列上，此时镜像队列处于不可用状态。（defalut setting）

当不可控条件下故障，机器断电损坏等，master将会转移到未同步完成的节点上。

Loss of a master while all slaves are stopped

通常情况下，我们希望最后的节点在重启后继续接管master权限，因为该情况下，最后的master节点保存了最全的信息，而这些信息其他之前故障的slave节点并不能接收到。

然而，我们在Stopping nodes and synchronisation一节中知道，新启动加入集群的slave会首先清空其持久化的消息信息，作为一个新节点加入集群，所以在这里我们需要先执行rabbitmqctl forget_cluster_node，此时rabbitmq会尝试寻找当前node上的所有queue的master节点机器，并在master机器再次重启加入集群后重新提升其为master节点。

master节点故障时：

一个slave被提升为新的master。被提升的slave是“最旧”的slave。因为该slave与原master中内容完全同步的几率最大。然而，也有可能所有的salve都未与master完全同步，此时只有master中存在而slaves中不存在的message将丢失。

slave认为所有之前的consumers的连接突然断开了。因此，它重新将已经投递但还未被确认的messages重新排队。这些“未被确认的”message可能包含client已发出确认但确认在到达master前丢失了的情形，也包含client已发出确认且确认已到达master但在master广播给slaves时丢失的情形。在上述任意一种情况下，新的master都必须为他认为未收到确认的message重新排队。

之前请求原master的clients被取消。

由于重新排队，从queue重新consume的clients需要知道此时可能接收到之前已经被处理过的message。

x-cancel-on-ha-failover=true时，consuming过程将被取消，consumer cancellation notification会被发出，由业务方确认是否重新消费。

当master节点故障时，如果publisher使用的时noAck模式，消息可能会丢失，因为noAck模式下，消息一旦进入broker的处理流程，就代表消息已经处理完成，不需要任何确认过程。此时master发生故障，将不会requeue该消息，消息有丢失的风险。

4.消息队列的应用场景举例

可靠的延迟队列

RPC:

Our RPC will work like this:

When the Client starts up, it creates an anonymous exclusive callback queue.

For an RPC request, the Client sends a message with two properties:replyTo, which is set to the callback queue andcorrelationId, which is set to a unique value for every request.

The request is sent to anrpc_queuequeue.

The RPC worker (aka: server) is waiting for requests on that queue. When a request appears, it does the job and sends a message with the result back to the Client, using the queue from thereplyTofield.

The client waits for data on the callback queue. When a message appears, it checks thecorrelationIdproperty. If it matches the value from the request it returns the response to the application.

参考资料：

https://www.cloudamqp.com/blog/2014-12-03-what-is-message-queuing.html

http://tech.meituan.com/mq-design.html

https://segmentfault.com/a/1190000003059871

http://www.liaoqiqi.com/post/215

http://www.rabbitmq.com/features.html

http://club.alibabatech.org/resource_detail.htm?topicId=61

https://segmentfault.com/a/1190000007123977

https://www.rabbitmq.com/tutorials/amqp-concepts.html

http://www.rabbitmq.com/confirms.html

http://blog.csdn.net/summerhust/article/details/46325107

http://dbaplus.cn/news-21-460-1.html