玩转不同业务场景，这些RabbitMQ特性会是得力助

我是在解决分布式事务的一致性问题时了解到的RabbitMQ&＃xff0c;当时主要是要基于RabbitMQ来实现我们分布式系统之间对有事务可靠性要求的系统间通信。

提到RabbitMQ&＃xff0c;不难想到的几个关键字&＃xff1a;消息中间件、消息队列。当时在大学学习操作系统这门课&＃xff0c;消息队列不难想到生产者消费者模式。

(PS&＃xff1a;操作系统这门课程真的很好也很重要&＃xff0c;其中的一些思想在我工作的很长一段时间内给了我很大帮助和启发&＃xff0c;给我提供了许多解决问题的思路。强烈建议每一个程序员都去学一学操作系统。)

消息中间件也可以称消息队列&＃xff0c;是指用高效可靠的消息传递机制进行与平台无关的数据交流&＃xff0c;并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息队列模型&＃xff0c;可以在分布式环境下扩展进程的通信。

当下主流的消息中间件有RabbitMQ、Kafka、ActiveMQ、RocketMQ等。其能在不同平台之间进行通信&＃xff0c;常用来屏蔽各种平台协议之间的特性&＃xff0c;实现应用程序之间的协同。优点在于能够在客户端和服务器之间进行同步和异步的连接&＃xff0c;并且在任何时刻都可以将消息进行传送和转发&＃xff0c;是分布式系统中非常重要的组件&＃xff0c;主要用来解决应用耦合、异步通信、流量削峰等问题。

消息中间件几大主要作用如下&＃xff1a;

• 解耦

• 冗余(存储)

• 扩展性

• 削峰

• 可恢复性

• 顺序保证

• 缓冲

• 异步通信

P2P模式

P2P模式包含三个角色&＃xff1a;消息队列&＃xff08;Queue&＃xff09;、发送者(Sender)、接收者(Receiver)。每个消息都被发送到一个特定的队列&＃xff0c;接收者从队列中获取消息。队列保留着消息&＃xff0c;直到它们被消费或超时。

P2P的特点&＃xff1a;

• 每个消息只有一个消费者&＃xff08;Consumer&＃xff09;&＃xff0c;即一旦被消费&＃xff0c;消息就不再在消息队列中

• 发送者和接收者之间在时间上没有依赖性&＃xff0c;也就是说当发送者发送了消息之后&＃xff0c;不管接收者有没有正在运行它不会影响到消息被发送到队列

• 接收者在成功接收消息之后需向队列应答成功

• 如果希望发送的每个消息都会被成功处理的话&＃xff0c;那么需要P2P模式

Pub/Sub模式

Pub/Sub模式包含三个角色&＃xff1a;主题&＃xff08;Topic&＃xff09;、发布者&＃xff08;Publisher&＃xff09;、订阅者&＃xff08;Subscriber&＃xff09; 。多个发布者将消息发送到Topic&＃xff0c;系统将这些消息传递给多个订阅者。

Pub/Sub的特点&＃xff1a;

• 每个消息可以有多个消费者

• 发布者和订阅者之间有时间上的依赖性。针对某个主题&＃xff08;Topic&＃xff09;的订阅者&＃xff0c;它必须创建一个订阅者之后&＃xff0c;才能消费发布者的消息

• 为了消费消息&＃xff0c;订阅者必须保持运行的状态

• 如果希望发送的消息可以不被做任何处理、或者只被一个消息者处理、或者可以被多个消费者处理的话&＃xff0c;那么可以采用Pub/Sub模型

Kafka

Kafka是LinkedIn开源的分布式发布-订阅消息系统&＃xff0c;目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费&＃xff0c;追求高吞吐量&＃xff0c;一开始的目的就是用于日志收集和传输。0.8版本开始支持复制&＃xff0c;不支持事务&＃xff0c;对消息的重复、丢失、错误没有严格要求&＃xff0c;适合产生大量数据的互联网服务的数据收集业务。

RabbitMQ

RabbitMQ是使用Erlang语言开发的开源消息队列系统&＃xff0c;基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由&＃xff08;包括点对点和发布/订阅&＃xff09;、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景&＃xff0c;对性能和吞吐量的要求还在其次。

RocketMQ

RocketMQ是阿里开源的消息中间件&＃xff0c;它是纯Java开发&＃xff0c;具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka&＃xff0c;但并不是Kafka的一个Copy&＃xff0c;它对消息的可靠传输及事务性做了优化&＃xff0c;目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。

RabbitMQ比Kafka可靠&＃xff0c;Kafka更适合IO高吞吐的处理&＃xff0c;一般应用在大数据日志处理或对实时性&＃xff08;少量延迟&＃xff09;&＃xff0c;可靠性&＃xff08;少量丢数据&＃xff09;要求稍低的场景使用&＃xff0c;比如ELK日志收集。

RabbitMQ是流行的开源消息队列系统&＃xff0c;是AMQP&＃xff08;高级消息队列协议&＃xff09;的标准实现。它支持多种客户端&＃xff0c;如&＃xff1a;Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等&＃xff0c;支持AJAX&＃xff0c;持久化&＃xff0c;用于在分布式系统中存储转发消息&＃xff0c;在易用性、扩展性、高可用性等方面表现不俗。

RabbitMQ是使用Erlang编写的一个开源的消息队列&＃xff0c;本身支持很多的协议&＃xff1a;AMQP&＃xff0c;XMPP, SMTP, STOMP&＃xff0c;也正是如此&＃xff0c;使的它变的非常重量级&＃xff0c;更适合于企业级的开发。它同时实现了一个Broker构架&＃xff0c;这意味着消息在发送给客户端时先在中心队列排队&＃xff0c;对路由(Routing)、负载均衡(Load balance)或者数据持久化都有很好的支持。

其主要特点如下&＃xff1a;

• 可靠性

• 灵活的路由

• 扩展性

• 高可用性

• 多种协议

• 多语言客户端

• 管理界面

• 插件机制

RabbitMQ从整体上来看是一个典型的生产者消费者模型&＃xff0c;主要负责接收、存储和转发消息。其整体模型架构如下图所示&＃xff1a;

我们先来看一个RabbitMQ的运转流程&＃xff0c;稍后会对这个流程中所涉及到的一些概念进行详细的解释。

生产者&＃xff1a;

• 生产者连接到RabbitMQ Broker&＃xff0c;建立一个连接(Connection)开启一个信道(Channel)

• 生产者声明一个交换器&＃xff0c;并设置相关属性&＃xff0c;比如交换机类型、是否持久化等

• 生产者声明一个队列井设置相关属性&＃xff0c;比如是否排他、是否持久化、是否自动删除等

• 生产者通过路由键将交换器和队列绑定起来

• 生产者发送消息至RabbitMQ Broker&＃xff0c;其中包含路由键、交换器等信息

• 相应的交换器根据接收到的路由键查找相匹配的队列

• 如果找到&＃xff0c;则将从生产者发送过来的消息存入相应的队列中

• 如果没有找到&＃xff0c;则根据生产者配置的属性选择丢弃还是回退给生产者

• 关闭信道

• 关闭连接

消费者&＃xff1a;

• 消费者连接到RabbitMQ Broker &＃xff0c;建立一个连接(Connection)&＃xff0c;开启一个信道(Channel)

• 消费者向RabbitMQ Broker请求消费相应队列中的消息&＃xff0c;可能会设置相应的回调函数

• 等待RabbitMQ Broker回应并投递相应队列中的消息&＃xff0c;消费者接收消息

• 消费者确认(ack) 接收到的消息

• RabbitMQ从队列中删除相应己经被确认的消息

• 关闭信道

• 关闭连接
  

信道

这里我们主要讨论两个问题&＃xff1a;

为何要有信道?

主要原因还是在于TCP连接的“昂贵”性。无论是生产者还是消费者&＃xff0c;都需要和RabbitMQ Broker 建立连接&＃xff0c;这个连接就是一条TCP连接。而操作系统对于TCP连接的创建与销毁是非常昂贵的开销。

假设消费者要消费消息&＃xff0c;并根据服务需求合理调度线程&＃xff0c;若只进行TCP连接&＃xff0c;那么当高并发的时候&＃xff0c;每秒可能都有成千上万的TCP连接&＃xff0c;不仅仅是对TCP连接的浪费&＃xff0c;也很快会超过操作系统每秒所能建立连接的数量。如果能在一条TCP连接上操作&＃xff0c;又能保证各个线程之间的私密性就完美了&＃xff0c;于是信道的概念出现了。

信道为何?

信道是建立在Connection之上的虚拟连接。当应用程序与Rabbit Broker建立TCP连接的时候&＃xff0c;客户端紧接着可以创建一个AMQP信道(Channel) &＃xff0c;每个信道都会被指派一个唯一的ID。RabbitMQ 处理的每条AMQP指令都是通过信道完成的。信道就像电缆里的光纤束&＃xff0c;一条电缆内含有许多光纤束&＃xff0c;允许所有的连接通过多条光线束进行传输和接收。

生产者消费者

关于生产者消费者我们需要了解几个概念&＃xff1a;

• Producer&＃xff1a;生产者&＃xff0c;即消息投递者一方。

• 消息&＃xff1a;消息一般分两个部分——消息体(payload)和标签。标签用来描述这条消息&＃xff0c;如&＃xff1a;一个交换器的名称或者一个路由Key&＃xff0c;Rabbit通过解析标签来确定消息的去向&＃xff0c;payload是消息内容可以使一个json、数组等等。

• Consumer&＃xff1a;消费者&＃xff0c;就是接收消息的一方。消费者订阅RabbitMQ的队列&＃xff0c;当消费者消费一条消息时&＃xff0c;只是消费消息的消息体。在消息路由的过程中&＃xff0c;会丢弃标签&＃xff0c;存入到队列中的只有消息体。

• Broker&＃xff1a;消息中间件的服务节点。

队列、交换器、路由key、绑定

从RabbitMQ的运转流程我们可以知道生产者的消息是发布到交换器上的。而消费者则是从队列上获取消息的。那么消息到底是如何从交换器到队列的呢&＃xff1f;我们先具体了解一下这几个概念&＃xff1a;

Queue&＃xff1a;队列&＃xff0c;是RabbitMQ的内部对象&＃xff0c;用于存储消息。RabbitMQ中消息只能存储在队列中&＃xff0c;生产者投递消息到队列&＃xff0c;消费者从队列中获取消息并消费。多个消费者可以订阅同一个队列&＃xff0c;这时队列中的消息会被平均分摊(轮询)给多个消费者进行消费&＃xff0c;而不是每个消费者都收到所有的消息进行消费。

注意&＃xff1a;RabbitMQ不支持队列层面的广播消费&＃xff0c;如果需要广播消费&＃xff0c;可以采用一个交换器通过路由Key绑定多个队列&＃xff0c;由多个消费者来订阅这些队列的方式。

Exchange&＃xff1a;交换器。在RabbitMQ中&＃xff0c;生产者并非直接将消息投递到队列中。真实情况是&＃xff0c;生产者将消息发送到Exchange(交换器)&＃xff0c;由交换器将消息路由到一个或多个队列中。如果路由不到&＃xff0c;或返回给生产者&＃xff0c;或直接丢弃&＃xff0c;或做其它处理。

RoutingKey&＃xff1a;路由Key。生产者将消息发送给交换器的时候&＃xff0c;一般会指定一个RoutingKey&＃xff0c;用来指定这个消息的路由规则。这个路由Key需要与交换器类型和绑定键(BindingKey)联合使用才能最终生效。在交换器类型和绑定键固定的情况下&＃xff0c;生产者可以在发送消息给交换器时通过指定RoutingKey来决定消息流向哪里。

Binding&＃xff1a;RabbitMQ通过绑定将交换器和队列关联起来&＃xff0c;在绑定的时候一般会指定一个绑定键&＃xff0c;这样RabbitMQ就可以指定如何正确的路由到队列了。

从这里我们可以看到在RabbitMQ中交换器和队列实际上可以是一对多&＃xff0c;也可以是多对多关系。交换器和队列就像我们关系数据库中的两张表。它们同归BindingKey做关联(多对多关系表)。在我们投递消息时&＃xff0c;可以通过Exchange和RoutingKey(对应BindingKey)就可以找到相对应的队列。

RabbitMQ主要有四种类型的交换器&＃xff1a;

fanout&＃xff1a;扇形交换器&＃xff0c;它会把发送到该交换器的消息路由到所有与该交换器绑定的队列中。如果使用扇形交换器&＃xff0c;则不会匹配路由Key。

direct&＃xff1a;direct交换器&＃xff0c;会把消息路由到RoutingKey与BindingKey完全匹配的队列中。

topic&＃xff1a;完全匹配BindingKey和RoutingKey的direct交换器有些时候并不能满足实际业务的需求。topic类型的交换器在匹配规则上进行了扩展&＃xff0c;它与direct类型的交换器相似&＃xff0c;也是将消息路由到BindingKey和RoutingKey相匹配的队列中&＃xff0c;但这里的匹配规则有些不同&＃xff0c;它约定:

• RoutingKey为一个点号"."分隔的字符串(被点号"."分隔开的每一段独立的字符串称为一个单词)&＃xff0c;如"hs.rabbitmq.client"&＃xff0c;"com.rabbit.client"等。

• BindingKey和RoutingKey一样也是点号"."分隔的字符串&＃xff1b;

• BindingKey中可以存在两种特殊字符串"*"和"#"&＃xff0c;用于做模糊匹配&＃xff0c;其中"*"用于匹配一个单词&＃xff0c;"#"用于匹配多规格单词(可以是零个)。

如图&＃xff1a;

• 路由键为" apple.rabbit.client" 的消息会同时路由到Queuel 和Queue2&＃xff1b;

• 路由键为" orange.mq.client" 的消息只会路由到Queue2 中&＃xff1b;

• 路由键为" apple.mq.demo" 的消息只会路由到Queue2中&＃xff1b;

• 路由键为" banana.rabbit.demo" 的消息只会路由到Queuel 中&＃xff1b;

• 路由键为" orange.apple.banana" 的消息将会被丢弃或者返回给生产者因为它没有匹配任何路由键。

header&＃xff1a;headers类型的交换器不依赖于路由键的匹配规则来路由消息&＃xff0c;而是根据发送的消息内容中的headers属性进行匹配。在绑定队列和交换器时制定一组键值对&＃xff0c;当发送消息到交换器时&＃xff0c;RabbitMQ会获取到该消息的headers (也是一个键值对的形式) &＃xff0c;对比其中的键值对是否完全匹配队列和交换器绑定时指定的键值对&＃xff0c;如果完全匹配则消息会路由到该队列&＃xff0c;否则不会路由到该队列。

注&＃xff1a;该交换器类型性能较差且不实用&＃xff0c;因此一般不会用到。

了解了上面的概念&＃xff0c;我们再来思考消息是如何从交换器到队列的。首先Rabbit在接收到消息时&＃xff0c;会解析消息的标签从而得到消息的交换器与路由Key信息。然后根据交换器的类型、路由Key以及该交换器和队列的绑定关系来决定消息最终投递到哪个队列里面。