本文主要是介绍架构设计:系统间通信——MQ:消息协议,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
1、概述
从本文开始,我们介绍另一类型的系统间通讯及输:MQ消息队列。首先我们将讨论几种常用消息队列协议的基本原理和工作方式,包括MQTT、XMPP、Stomp、AMQP、OpenWire等。然后在这个基础上介绍两款MQ产品:ActiveMQ和RabbitMQ,它们是现在业务系统中应用广泛的消息队列软件。包括他们的安装、运行、支持协议、集群化和调用方式。
当然,在这个过程中我们还会提到其他的消息队列协议(或者实现),例如微软JBossMQ、MSMQ、商业化产品WebSphere MQ、Oracle高级队列(AQ)等。我们还会讨论这些眼花缭乱的协议、软件、程序库之间的关系。
随后我们会花一些篇幅,讨论现在新兴的消息队列Kafka和ZeroMQ。它们的应用越来越广泛,尤其在大数据的采集方面。最后我们将使用消息队列搭建一个高性能的日志采集系统,作为实战。
2、基本概念
2-1、消息
首先有三个基本概念在开篇前我们需要进行讨论:消息、消息协议、消息队列。消息既是信息的载体 这个描述相信各位读者都能够明白。为了让消息发送者和消息接收者都能够明白消息所承载的信息(消息发送者需要知道如何构造消息;消息接收者需要知道如何解析消息),它们就需要按照一种统一的格式描述消息,这种统一的格式称之为消息协议。所以,有效的消息一定具有某一种格式;而没有格式的消息是没有意义的。
而消息从发送者到接收者的方式也有两种。一种我们可以称为即时消息通讯,也就是说消息从一端发出后(消息发送者)立即就可以达到另一端(消息接收者),这种方式的具体实现就是我们已经介绍过的RPC(当然单纯的http通讯也满足这个定义);另一种方式称为延迟消息通讯,即消息从某一端发出后,首先进入一个容器进行临时存储,当达到某种条件后,再由这个容器发送给另一端。 这个容器的一种具体实现就是消息队列。
2-2、知识结构
消息队列和已经介绍过的RPC相同的是:无论是RPC也好,消息队列也好他们都建立在网络IO模型基础上(我们已经介绍过多种网络IO模型)。先进的网络IO模型将赋予MQ协议优异的性能表现(当然,性能也不仅仅取决于网络IO模型)。
从上图可以看到,某一种消息通讯软件(或者叫做程序库)的实现都建立在“协议”基础上:RMI程序库建立在RMI协议上(RMI协议是JAVA规范协议的一部分) ,属于一种“即时消息通讯”;RabbitMQ和Qpid消息通讯软件的设计依据是AMQP协议,属于一种“延迟消息通讯”。
虽然消息协议存在“私有协议”和“开放协议”之分(是否向行业开放消息规范文档、是否允许某个组织更改协议),虽然某一个软件(程序库)不一定只支持一种协议(例如ActiveMQ实现了多种消息协议),虽然某一种协议也不一定只有一种软件(程序库)实现(例如能够支持webservice协议的程序库就有Codehaus XFire、Apache CXF、Jboss RESTEasy等),但是这并不影响“某一种消息通讯软件(或者叫做程序库)的实现都建立在“协议”基础上”的概念,反而是这个基本概念加强了。
3、消息协议
那么要理解消息队列,我们就应该从这些支持“延迟消息通讯”的消息协议开始讨论。这个小节我们首先为各位读者介绍几种使用的消息协议,他们是XMPP、Stomp和AMQP。为了承接后文我们讲解的MQ软件,这三个协议中我们又着重讲解AMQP协议。
3-1、XMPP协议
3-1-1、定义
XMPP is the Extensible Messaging and Presence Protocol, a set of open technologies for instant messaging, presence, multi-party chat, voice and video calls, collaboration, lightweight middleware, content syndication, and generalized routing of XML data.
以上内容引用自XMPP官网,这个定义已经可以清楚表明XMPP协议的用途和特性。XMPP的前身是Jabber,一个开源形式组织制定的网络即时通信协议。XMPP目前被IETF国际标准组织完成了标准化工作。
XMPP基于XML,用于IM系统的开发。国内比较流行的XMPP服务器叫做Openfire,它使用MINA作为下层的网络IO框架(不是MINA2是MINA1);国外用的比较多的XMPP服务器叫做Tigase,它的官网号称单节点可以支撑50万用户在线,集群可以支持100万用户在线:(http://projects.tigase.org/)
Cluster with over 1mn online users . 500k online users on a single machine
当然如果读者所在公司需要开发IM系统,除了使用现成的XMPP服务器以外,还需要实现了XMPP协议的客户端或者开发包(以便进行扩展开发)。您可以在XMPP官网查看到XMPP官方推荐的开发包,各种语言的支持基本上都有:https://xmpp.org/software/libraries/
笔者曾参与过某几款IM系统的开发(包括自己创业的项目),总的来说XMPP协议本身是不错的选择,但是学习起来会耗费相当的时间,并且某些XMPP客户端、服务器端或者程序库并没有这些开发团队宣传的那么稳定好用。所以如果您的公司需要进行IM系统的开发,那么创立私有的消息协议也会是一个不错的选择。
3-1-2、协议通讯过程示例
为了让各位读者对XMPP协议有一个感性认识,这里我们给出一个XMPP协议处理“IM用户登录”操作的过程(XMPP的登录方式分为有用户密码和无用户密码两种方式,这里我们介绍无密码登录方式)。
XMPP协议本身细节比较丰富,这里我们只讨论登录操作,如果读者有兴趣可以下载全套的XMPP官方规范文档进行研究(XMPP | The universal messaging standard):
通过上图可以看到,XMPP协议中的xml片段。这里出现了几个XMPP协议中的关键信息,例如:
-
stream标记:通讯流标记,是指XMPP的客户端或者服务器端向对方发起的通讯请求(或者响应)。通讯流并不携带正真的内容信息,指示表明客户端和服务器端发生了一次交互。stream的属性包括:to、from、id、xml:lang、version等。
-
iq标记:iq标记是Info/Query的简称(你可以理解成查询信息请求),一般是一组的形式出现,由客户端发起查询请求,由服务器端返回查询结果。由于查询请求的类型不一样,iq标记中可以嵌入的子标记就有很多。例如,可以嵌入bind标记,表明某个用户和jid的绑定关系;可以嵌入多个item标记,表明查询得到的这个用户的好友信息(如下)。
<iq to='somenode@example.com/someresource' type='result' id='roster'> <query xmlns='jabber:iq:roster'> <item jid='friend1@example.com' name='someone1'/> <item jid='friend2@example.com' name='someone2'/> </query>
</iq>
- jid标记:jid(JabberID)是XMPP协议中标示,它用来标示XMPP网络中的各个XMPP实体(实体可以是某一个用户、某一个服务器、某一个聊天室),规范格式如下:
jid = [ node "@" ] domain [ "/" resource ]
- 还有未出现的message、presence标记:message是实体内容标记,记录了聊天的真实内容;presence标记表示了XMPP用户的服务状态(离线,在线、忙碌等)。示例如下:
<message to="somenode@example.com/someresource" type="chat"> <body>helloword。。。</body>
</message>
3-2、Stomp协议
3-2-1、定义
Stomp协议,英文全名Streaming Text Orientated Message Protocol,中文名称为 ‘流文本定向消息协议’。是一种以纯文本为载体的协议(以文本为载体的意思是它的消息格式规范中没有类似XMPP协议那样的xml格式要求,你可以将它看作‘半结构化数据’)。目前Stomp协议有两个版本:V1.1和V1.2。
一个标准的Stomp协议包括以下部分:命令/信息关键字、头信息、文本内容。如下图所示:
以下为一段简单的协议信息示例:
CONNECT
accept-version:1.2
someparam1:value1
someparam2:value2this is conntecon ^@
上面的示例中,我们使用了Stomp协议的CONNECT命令,它的意思为连接到Stomp代理端,并且携带了要求代理端的版本信息和两个自定义的K-V信息(请注意’^@’符号,STOMP协议中用它来表示NULL)。
Stomp协议中有两个重要的角色:STOMP客户端与任意STOMP消息代理(Broker)。如下图所示:
看了上面的示意图后有的读者可能会问:为什么称为Stomp消息代理,而不称为Stomp消息服务?因为Stomp Broker只是负责接受和存储客户端发来的消息、只是按照客户端要求的路径转发消息,只是管理客户端连接和订阅:它并不负责根据消息内容做任何业务处理。所以将它称为消息代理端更贴切。
由于Stomp协议的结构如此简单,以至于任何理解Stomp协议命令格式的技术人员都可以开发Stomp的代理端或者Stomp的客户端,并将自己满足Stomp协议的系统轻松接入另一个同样满足Stomp协议的第三方系统(例如activeMQ)。
3-2-2、基本命令/返回信息
和介绍XMPP协议的方式类似,为了让读者对Stomp协议有进一步的认识,本小节我们介绍Stomp协议的基本命令和代理端返回的信息种类,并且列举一些实例进行使用讲解。
在Stomp协议中,主要有以下命令/返回信息(有的文章中也称一个完整的信息为帧)。这些命令/返回信息构成了Stomp协议的主体,并能够支持您的Stomp客户端和Stomp代理端完成连接、发送、订阅、事务、响应的整个操作过程。这些命令/返回是:
-
CONNECT/STOMP命令: 客户端通过使用CONNECT命令,连接到Stomp代理端。如果使用STOMP命令,那么Stomp代理端的版本必须是1.2。
-
CONNECTED信息:当Stomp代理端收到客户端发送来的Connect命令并且处理成功后,将向这个客户端返回CONNECTED状态信息;如果这个过程中出现任何问题,还可能返回ERROR信息
-
SEND 发送命令:客户端使用SEND命令,向某个指定位置(代理端上的一个虚拟路径)发送内容。这样在这个路径上订阅了消息事件的其它客户端,将能够收到这个消息。
-
SUBSCRIBE 订阅命令:客户端使用SUBSCRIBE订阅命令,向Stomp服务代理订阅某一个虚拟路径上的监听。这样当其它客户端使用SEND命令发送内容到这个路径上时,这个客户端就可以收到这个消息。在使用SUBSCRIBE时,有一个重要的ACK属性。这个ACK属性说明了Stomp服务代理端发送给这个客户端的消息是否需要收到一个ACK命令,才认为这个消息处理成功了。如下所示:
SUBSCRIBE
id:XXXXXXXXX
destination:/test
ack:client^@
以上SUBSCRIBE命令信息说明,客户端订阅的虚拟位置是test。且命令信息中ack属性为client,说明当客户端收到消息时,必须向代理端发送ack命令,代理端才认为这个消息处理成功了(ack的值只有三种:auto(默认)、client和client-individual)。
-
UNSUBSCRIBE 退订命令:客户端使用这个命令,取消对某个路径上消息事件的监听。如果客户端给出的路径之前就没有被这个客户端订阅,那么这个命令执行无效。
-
MESSAGE 信息:当客户端在某个订阅的位置收到消息时,这个消息将通过MESSAGE关键字进行描述。类似以下信息就是从代理端拿到的消息描述:
MESSAGE
redelivered:true
message-id:ID:localhost-34450-1457321490460-4:24:-1:1:1
destination:/test
timestamp:1457331607873
expires:0
priority:42345431457331607861
-
BEGIN 开始事务命令: Stomp协议支持事务模式,在这种模式下,使用Send命令从某个客户端发出的消息,在没有使用COMMIT正式提交前,这些消息是不会真正发送给Stomp代理端的。BEGIN命令就是用于开启事务。注意,一个事务中可以有一条消息,也可以有多条消息。
-
COMMIT 提交命令: 当完成事务中的信息定义后,使用该命令提交事务。只有使用COMMIT命令后,在某一个事务中的一条或者多条消息才会进入Stomp代理端的队列(订阅了事件的其它客户端才能收到这些消息)。
-
ABORT 取消/终止事务命令:很明显,这个命令用于取消/终止当前还没有执行COMMIT命令的事务。
-
ACK 确认命令:当客户端使用SUBSCRIBE命令进行订阅时,如果在SUBSCRIBE命令中制定ack属性为client,那么这个客户端在收到某条消息(id为XXXX)后,必须向Stomp代理端发送ACK命令,这样代理端才会认为消息处理成功了;如果Stomp客户端在断开连接之前都没有发送ACK命令,那么Stomp代理端将在这个客户端断开连接后,将这条消息发送给其它客户端。
ACK
id:MESSAGE ID^@
请注意head部分的id属性,传递的id属性是之前收到的MESSAGE信息的id标示。
-
NACK 不确认命令:同样是以上的SUBSCRIBE命令的状态下,如果这时Stomp客户端向Stomp代理端发送NACK信息,证明这条消息在这个客户端处理失败。Stomp代理端将会把这条消息发送给另一个客户端(无论当前的客户端是否断开连接)。
-
DISCONNECT 断开命令:这个命令将断开Stomp客户端与Stomp代理端的连接。
上篇文章中我们重点讨论了“协议”的重要性,并为各位读者介绍了Stomp协议和XMPP协议。这两种协议是消息队列中两种不同使用场景下的典型代表。本文主要接续上文的篇幅,继续讨论消息队列中另一种典型协议:AMQP协议。
3-3、AMQP协议
AMQP协议的全称是:Advanced Message Queuing Protocol(高级消息队列协议)。目前AMQP协议的版本为 Version 1.0,这个协议标准在2014年通过了国际标准组织 (ISO) 和国际电工委员会 (IEC) 的投票,成为了新的 ISO 和 IEC 国际化标准。目前支持AMQP的软件厂商包括:
3-3-1、协议概览
在网络上讲解AQMP协议的文章已经有很多了,您可以在百度、Google、必应上搜索关键字‘AMQP’,就会出现很多相关文章。虽然文章数量比较多,但是却鲜有质量过硬的文章(当然除了AMQP官网 Home | AMQP 的协议说明文档)。本小节的内容我试图在能力所及的范围内,为各位读者将AMQP协议的核心要点讲清楚。为了达到这个目的,首先将AMQP协议的原理用下图进行一个全面呈现,然后在详细讲解图中的每一个要点:
从上图我们可以看到AMQP协议的各个组成部分:
-
AMQP协议中的元素包括:Message(消息体)、Producer(消息生产者)、Consumer(消息消费者)、Virtual Host(虚拟节点)、Exchange(交换机)、Queue(队列)等
-
由Producer(消息生产者)和Consumer(消息消费者)构成了AMQP的客户端,他们是发送消息和接收消息的主体。AMQP服务端称为Broker,一个Broker中一定包含完整的Virtual Host(虚拟主机)、 Exchange(交换机)、Queue(队列)定义。
-
一个Broker可以创建多个Virtual Host(虚拟主机),我们将讨论的Exchange和Queue都是虚拟机中的工作元素(还有User元素)。注意,如果AMQP是由多个Broker构成的集群提供服务,那么一个Virtual Host也可以由多个Broker共同构成。
-
Connection是由Producer(消息生产者)和Consumer(消息消费者)创建的连接,连接到Broker物理节点上。但是有了Connection后客户端还不能和服务器通信,在Connection之上客户端会创建Channel,连接到Virtual Host或者Queue上,这样客户端才能向Exchange发送消息或者从Queue接受消息。一个Connection上允许存在多个Channel,只有Channel中能够发送/接受消息。
-
Exchange元素是AMQP协议中的交换机,Exchange可以绑定多个Queue也可以同时绑定其他Exchange。消息通过Exchange时,会按照Exchange中设置的Routing(路由)规则,将消息发送到符合的Queue或者Exchange中。
那么AMQP消息在这个结构中是如何通过Producer发出,又经过Broker最后到达Consumer的呢?请看下图:
-
在Producer(消息生产者)客户端建立了Channel后,就建立了到Broker上Virtual Host的连接。接下来Producer就可以向这个Virtual Host中的Exchange发送消息了。
-
Exchange(交换机)能够处理消息的前提是:它至少已经和某个Queue或者另外的Exchange形成了绑定关系,并设置好了到这些Queue和Excahnge的Routing(路由规则)。Excahnge中的Routing有三种模式,我们随后会讲到。在Exchange收到消息后,会根据设置的Routing(路由规则),将消息发送到符合要求的Queue或者Exchange中(路由规则还会和Message中的Routing Key属性配合使用)。
-
Queue收到消息后,可能会进行如下的处理:如果当前没有Consumer的Channel连接到这个Queue,那么Queue将会把这条消息进行存储直到有Channel被创建(AMQP协议的不同实现产品中,存储方式又不尽相同);如果已经有Channel连接到这个Queue,那么消息将会按顺序被发送给这个Channel。
-
Consumer收到消息后,就可以进行消息的处理了。但是整个消息传递的过程还没有完成:视设置情况,Consumer在完成某一条消息的处理后,将需要手动的发送一条ACK消息给对应的Queue(当然您可以设置为自动发送,或者无需发送)。Queue在收到这条ACK信息后,才会认为这条消息处理成功,并将这条消息从Queue中移除;如果在对应的Channel断开后,Queue都没有这条消息的ACK信息,这条消息将会重新被发送给另外的Channel。当然,您还可以发送NACK信息,这样这条消息将会立即归队,并发送给另外的Channel。
3-3-2、Message(消息体)
通过上一小节的描述,我们可以看到AMQP协议中消息的处理规则和Stomp协议中消息的处理规则有类似之处,比如对ACK、NACK的使用。但明显不同的地方还是很多,例如AMQP中Exchange元素提供的Routing路由规则,这显然比Stomp协议中直接发送给Queue要灵活得多。
为了支持AMQP协议中的这些规则,AMQP协议中的消息也必须有特定的格式,实际上AMQP协议要比Stomp协议复杂得多。下面我们就根据ISO/IEC发布的AMQP Version 1.0标准文档,来讨论一下AMQP协议中的消息格式。
首先要说明的是目前国内多个技术站点,详细介绍AMQP消息格式的文章本来就不多(不包括那些聊聊几笔的转发),而且基本上都没有详细讲解格式本身,只是粗略说明了AMQP消息采用二进制格式(任何应用层协议在网络上进行传输,都是使用二进制流进行的,所以这个说法当然没错)。
有的文章还向读者传递了错误的信息。例如说AMQP消息格式包括两部分:消息头和消息正文。 这是完全错误的,虽然AMQP消息格式确实包括Header和Body部分,但是绝对不止这两个部分。(如果真是这样,ISO/IEC组织就不需要使用125页的文档篇幅来进行说明了)
首先我们需要说明的是,作为一种网络通讯协议,AMQP工作在七层/五层网络模型的应用层,是一个典型的应用层协议;另外,由于AMQP协议存在多种元素定义,且这些元素定义工作在不同的领域。例如Channel的定义是为了基于网络连接记录会话状态;Queue等元素帮助AMQP完成路由规则,这些元素在Message消息记录中都需要有所体现。
所以AMQP协议首先要记录网络状态和会话状态,格式如下(AMQP帧的定义在《OASIS Advanced Message Queueing Protocol
(AMQP) Version 1.0》文档的第38页):
其中非PAYLOAD部分,在网络协议的应用层说明Channel的工作状态(当然还有说明整个AMQP消息的长度区域:SIZE),我们真正需要的内容存在PAYLOAD区域。PAYLOAD区域(译文称为‘交付区’)的格式如下(可以在《OASIS Advanced Message Queueing Protocol
(AMQP) Version 1.0》文档的第3部分:messaging第82页找到详细说明):
在PAYLAOD区域一共包含7个数据区域:header、delivery-annotations、message-annotations、properties、application-properties、application-data、footer。这些元素的作用如下:
-
header:header部分记录了AMQP消息的在‘支持AMQP的中间件’中的交互状态。例如该条消息在节点间被交互的总次数、优先级、TTL(Time To Live)值等信息。
-
delivery-annotations:在header部分只能传递规范的、标准的、经过ISO/IEC组织定义的属性。那么如果需要在header部分传递一些非标准信息怎么办呢?这就是delivery-annotations数据区域存在的意义:用来记录那些’非标’的header信息。
-
message-annotations:这个数据区域,用于存储一些自定义的辅助属性。和delivery-annotations区域的非标准信息不同,这里的自定义属性主要用于消息的转换。例如AMQP-JMS信息转换过程中将依据这个数据区域的“x-opt-jms-type”、“x-opt-to-type”、“x-opt-reply-type”和“name”属性进行JMS规范中对应的“JMSType”、“Type of the JMSDestination”、“Type of the JMSReplyTo”和“JMS_AMQP_MA_name”属相的转换。
-
properties:从整个AMQP消息的properties属性开始,到AMQP消息的application-data部分结束,才是AMQP消息的正文内容(译文称为‘裸消息’)。Properties属性记录了AMQP消息正文的那些‘不可变’属性。在properties部分只能传递规范的、标准的、经过ISO/IEC组织定义的属性。例如:消息id、分组id、发送者id、内容编码等。以下是AMQP协议文档中对Properties部分属性的描述(只能包含这些信息):
<type name="properties" class="composite" source="list" provides="section"><descriptor name="amqp:properties:list" code="0x00000000:0x00000073"/><field name="message-id" type="*" requires="message-id"/><field name="user-id" type="binary"/><field name="to" type="*" requires="address"/><field name="subject" type="string"/><field name="reply-to" type="*" requires="address"/><field name="correlation-id" type="*" requires="message-id"/><field name="content-type" type="symbol"/><field name="content-encoding" type="symbol"/><field name="absolute-expiry-time" type="timestamp"/><field name="creation-time" type="timestamp"/><field name="group-id" type="string"/><field name="group-sequence" type="sequence-no"/><field name="reply-to-group-id" type="string"/>
</type>
-
application-properties:‘应用数据’属性,在这部分数据中主要记录和应用有关的数据,AMQP的实现产品(例如RabbitMQ)需要用这部分数据决定其处理逻辑。例如:送入哪一个Exchange、消息的Routing值是什么、是否进行持久化等。
-
application-data:使用二进制格式描述的AMQP消息的用户部分内容。既是我们发送出去的真实内容。
-
footer:一般在这个数据区域存储辅助内容,例如消息的哈希值,HMAC,签名或者加密细节。
以上才是一个AMQP消息的完整结构。当然由于篇幅限制,在某一个数据区域的‘标准’属性就没有再细讲了,例如Properties数据区域定义的creation-time属性、Header数据区域定义的durable属性。有兴趣的朋友可以参考《OASIS Advanced Message Queueing Protocol
(AMQP) Version 1.0》,这个文档我已经上传到我的下载列表中,供大家免费下载^_^(OASISAdvancedMessageQueueingProtocol(AMQP)Version1.0-其它文档类资源-CSDN下载)。
3-3-3、Exchange(交换机)路由规则
Exchange交换机在AMQP协议中主要负责按照一定的规则,将收到的消息转发到已经和它事先绑定好的Queue或者另外的Exchange中。Excahnge交换机的这个处理过程称之为Routing(路由)。目前流行的AMQP路由实现一共有三种:分别是Direct Exchange、Fanout Exchange和Topic Exchange。需要特别注意的是:Exhange需要具备怎样的‘路由’规则,并没有在AMQP标准协议进行强行规定,目前流行的AMQP转发规则都是AMQP实现产品自行开发的(这也是为什么AMQP消息中和路由、过滤规则相关的属性是存放在application-properties区域的原因)。
A、Direct路由
direct模式从字面上的理解应该是‘引导’、‘直接’的含义。direct模式下Exchange将使用AMQP消息中所携带的Routing-Key和Queue中的Routing Key进行比较。如果两者完全匹配,就会将这条消息发送到这个Queue中。如下图所示:
B、Fanout路由
Fanout路由模式不需要Routing Key。当设置为Fanout模式的Exchange收到AMQP消息后,将会将这个AMQP消息复制多份,分别发送到和自己绑定的各个Queue中。如下图所示:
C、Topic路由
Topic模式是Routing Key的匹配模式。Exchange将支持使用‘#’和‘ * ’通配符进行Routing Key的匹配查找(‘#’表示0个或若干个关键词,‘ * ’表示一个关键词,注意是关键词不是字母)。如下图所示:
为了方便各位读者的理解,这里我们再举几个通配符匹配的示例:
-
“param.#”,可以匹配“param”、“param.test”、“param.value”、“param.test.child”等等AMQP消息的Routing Key;但是不能匹配诸如“param1.test”、“param2.test”、“param3.test”。以为param这个关键词和param1这个关键词不相同。
-
“param.*.* ”,可以匹配“param.test.test”、“param.test.value”、“param.test.child”等等AMQP消息的Routing Key;但是不能匹配诸如“param”、“param.test”、“parm.child”等等Routing Key。
-
“param.*.value”,可以匹配“param.value.value”、“param.test.value”等Routing Key;但是不能匹配诸如“param.value”、“param.value.child”等Routing Key。
注意,以上介绍的Direct 路由模式和Topic 路由模式中,如果Exchange交换机没有找到任何匹配Routing Key的Queue,那么这条AMQP消息会被丢弃。(只有Queue有保存消息的功能,但是Exchange并不负责保存消息)
4、不得不提的JMS规范
JMS不是消息队列,更不是某种消息队列协议。**JMS是Java消息服务接口,是一套规范的JAVA API 接口。这套规范接口由SUN提出,并在2002年发布JMS规范的Version 1.1版本。**JMS和消息中间件厂商无关,既然是一套接口规范,就代表这它需要各个厂商进行实现。好消息是,大部分消息中间件产品都支持JMS 接口规范。也就是说,您可以使用JMS API来连接Stomp协议的产品(例如ActiveMQ)。就像您可以使用JDBC API来连接ORACLE或者MYSQL一样。
部分网络上的资料都介绍JMS是一个消息队列,这个说法是错误的,会误导读者。难道你能说JDBC是数据库?
当然,这些具体实现JMS规范的JAVA API都是由具体的中间件厂商提供的。下面一段代码演示了如何使用JMS建立与ActiveMQ的连接:
package com.yinwenjie.test.testActivemq.jms;import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;/*** 测试使用JMS API连接ActiveMQ* @author yinwenjie*/
public class JMSProducer {/*** 由于是测试代码,这里忽略了异常处理。* 正是代码可不能这样做* @param args* @throws RuntimeException*/public static void main (String[] args) throws Exception {// 定义JMS-ActiveMQ连接信息ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616", "username", "password");Session session = null;Destination sendQueue;Connection connection = null;//进行连接connection = connectionFactory.createConnection();connection.start();//建立会话session = connection.createSession(true, Session.SESSION_TRANSACTED);//建立queue(当然如果有了就不会重复建立)sendQueue = session.createQueue("");//建立消息发送者对象MessageProducer sender = session.createProducer(sendQueue);TextMessage outMessage = session.createTextMessage();outMessage.setText("这是发送的消息内容");//发送(JMS是支持事务的)sender.send(outMessage);session.commit();//关闭sender.close();connection.close();connectionFactory.close();}
}
这里,再给各位读者一个官方文档。这个官方文档用于描述ActiveMQ消息中间件中实现的AMQP协议信息转换为JMS服务接口能够识别的数据信息(请仔细理解这句话黑体字部分的描述)。http://activemq.apache.org/amqp.html
5、后文介绍
通过两篇文章的篇幅,我们介绍了典型的消息队列协议。当然还有很多具体的消息队列协议没有讲到,但是通过介绍XMPP、AMQP、Stomp协议可以起到一个管中窥豹可见一斑的效果。另外我们还说明了JMS规范的具体含义,以便帮助读者纠正一些不正确的观点。
下一篇文章开始,我们将讲解两个典型的消息队列中间件:ActiveMQ和RabbitMQ。最后我们还会列举一个实际场景,然后通过消息队列技术一起搭建一个高性能的业务处理方案。
这篇关于架构设计:系统间通信——MQ:消息协议的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!