消息中间件

Posted 技术专家

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了消息中间件相关的知识，希望对你有一定的参考价值。

消息中间件选型

 

 

一、     分布式系统消息通信技术简介

分布式系统消息通信技术主要包括以下几种：

  1. 1.         RPC(Remote Procedure Call Protocol).

一般是C/S方式，同步的，跨语言跨平台，面向过程

  1. 2.         CORBA(Common Object Request Broker Architecture).

CORBA从概念上扩展了RPC。面向对象的，企业级的(面向对象中间件还有DCOM)

  1. 3.         RMI(Remote Method Invocation).

面向对象方式的 Java RPC

  1. 4.         WebService.

基于Web，C/S或B/S，跨系统跨平台跨网络。多为同步调用, 实时性要求较高

  1. 5.         MOM(Message oriented Middleware).

面向消息中间件，主要适用于消息通道、消息总线、消息路由和发布/订阅的场景。目前主流标准有JMS(Java Message Service)、AMQP(Advanced Message Queuing Protocol)和STOMP(Streaming Text Oriented Messaging Protocol)。

JMS是Java平台上的面向接口的消息规范，是一套API标准，并没有考虑异构系统。AMQP是一个面向协议的，跟语言平台无关的消息传递应用层协议规范。STOMP是流文本定向消息协议，是一种为MOM设计的简单文本协议。AMQP和STOMP都是跟http处于同一层的协议。

在 AMQP 模型中，消息的 producer 将 Message 发送给 Exchange，Exchange 负责交换 / 路由，将消息正确地转发给相应的 Queue。消息的 Consumer 从 Queue 中读取消息。

AMQP 系统构架

 

 

二、     常见开源跨平台MQ简介

目前业界上关于消息中间件的实现多达好几十种，可谓百花齐放，所用的实现语言同样也五花八门。下面挑选了一部分，在网上开源社区相对容易搜索出来的十多种MQ来作简单介绍。

开源MQ

概述

1.Qpid

Apach的一个开源AMQP实现，broker架构，有C++和Java两个版本

2.RabbitMQ

LShift 用Erlang实现，支持多协议，broker架构，重量级

3.ZeroMQ

AMQP最初设计者iMatix公司实现，轻量消息内核，无broker设计。C++实现

4.Jafka/Kafka

LinkedIn用Scala语言实现，支持hadoop数据并行加载

5.ActiveMQ

Apach的一种JMS具体实现，支持代理和p2p部署。支持多协议。Java实现

6.Apollo

ActiveMQ的下一代产品，支持多协议，Scala实现

7.Redis

Key-value  NoSQL数据库，有MQ的功能

8.MemcacheQ

国人利用memcache缓冲队列协议开发的消息队列,C/C++实现

9.Open-MQ

C++和QT实现，支持JMS

10.ActiveMQ-CPP

ActiveMQ的C++纯客户端库，用于跟ActiveMQ通信

11.MQ4CPP

一个C++实现的MQ，信息甚少

12.MetaQ

Alibaba对Kafka的改造，增加事务支持等新特性，用纯Java实现

13.Beanstalkd

一个类memcached协议设计的消息队列,C/C++实现

14.OpenAMQ

iMatix公司AMQP1.0的实现，类似rabbitMQ。C++实现。2010年项目放弃

15.Spread Toolkit

高性能的分布式分组消息系统，C++实现

16.SAFMQ

C++实现的储存转发消息队列中间件

17. Mosquitto

一个轻量级的IBM物联网连接协议的消息中间件实现，C/C++实现

18.MUSCLE

提供一个多路消息服务器和消息对象传递功能，支持C/C++

19.JORAM

一个类似OpenJMS(Sun OpenMQ)的JMS消息中间件，JAVA实现

 

Qpid

 

Qpid 是 Apache 开发的一款面向对象的消息中间件，它是一个 AMQP 的实现，可以和其他符合 AMQP 协议的系统进行通信。Qpid 提供了 C++/Python/Java/C# 等主流编程语言的客户端库，Qpid 提供了很多额外的 HA 特性，非常适于集群环境下的消息通信。

它提供了 C++ 和 Java 两个版本的 broker服务端，并支持多种语言的客户端。C++版本的服务器端具备高性能/低消耗以及RDMA支持；而Java版本的服务器则支持JMS。Qpid 还提供了一些额外的特性：

  • 采用 Corosync 来保证了集群环境下的 Fault-tolerant 特性
  • 支持 XML 类型的 Exchange，当消息格式为 XML 时，可以利用 Xquery 进行过滤
  • 支持 plugin，用户可以方便地增加新的功能，比如新的 exchange 类型

提供了安全认证特性，任何 producer/consumer 需要和 broker 通信时，都需要提供身份认证。QPID 的安全认证使用 SSL 协议。

授权协议： Apache

开发语言： Java C/C++

操作系统： 跨平台

官网：http://qpid.apache.org  最新版本0.30发布于2014-09-26。

 

有新浪的朋友比较推荐Qpid，它比rabbitmq要轻型，比zeromq保险点！各方面的文档都比较健全。目前在openstack中作为一种可选的消息中间件服务配置。MB(WSO2 Message Broker) 基于Apache Qpid，这是一个Java项目。开源的企业服务总线(ESB) – Celtix，基于Apache Incubator项目Qpid。

跟Qpid有关联的其他项目主要有:

ActiveMQ and AMQP using Qpid Proton

Apache Camel and AMQP using Qpid

Apache Axis and AMQP using Qpid

Microsofe Azure using Qpid Proton

Apache软件基金会一级开源项目入选Qpid is a top-level project at Apache

Pulp 2.5程序库内容管理工具里集成有Qpid.

Fedora默认使用Qpid作为消息队列

 

 

RabbitMQ

 

LShift 用Erlang编写的一个开源的消息队列，支持很多的协议：AMQP，XMPP, SMTP, STOMP，重量级，更适合于企业级的开发。代理(Broker)架构，对路由(Routing)，负载均衡(Load balance)或者数据持久化都有很好的支持。

缺点：可扩展性差，速度较慢，因为中央节点增加了延迟，消息封装后也比较大。

AMQP 里主要要说两个组件：Exchange 和 Queue （在 AMQP 1.0 里还会有变动），如下图所示，绿色的 X 就是 Exchange ，红色的是 Queue ，这两者都在 Server 端，又称作 Broker ，这部分是 RabbitMQ 实现的，而蓝色的则是客户端，通常有 Producer 和 Consumer 两种类型：

 

授权协议： MPL

开发语言： ErLang

操作系统： 跨平台

官网：http://www.rabbitmq.com/，最新版本3.4.3发布于2015-1-7

 

ØMQ(ZeroMQ)

 

早期需要设计可靠消息系统比如AMQP，但是这种方式引入了single-point broker。对于需要这种可靠消息系统的应用来说，需要在broker上面做相当多的事情确保可靠性以及性能。但是这样对于中小应用陷入了尴尬，为了使用这种方便的消息系统他们需要引入broker这么东西是不能够忍受的。我们需要的一种简单方便的消息传输系统，没有任何附加代价(比如所有数据都流经 broker)，这就是ZeroMQ设计初衷。

2010年3月30日，AMQP的最初设计者iMatix公司的首席执行官Pieter Hintjens宣布iMatix将退出AMQP工作组，而且为了简单得多，快的多的ZeroMQ，将不支持可能发布的AMQP/1.0。一个非常轻量级的消息内核，专门为高吞吐量/低延迟的场景开发。ZeroMQ支持许多高级消息场景，但是你必须实现ZeroMQ框架中的各个块（比如Socket或Device等）。没有中间件架构，应用程序端点扮演了这个服务角色。部署简单，仅提供非持久性的队列。与RabbitMQ相比，ØMQ并不像是一个传统意义上的消息队列服务器，事实上，它也根本不是一个服务器，它更像是一个底层的网络通讯库，在socket API之上做了一层封装，将网络通讯、进程通讯和线程通讯抽象为统一的API接口。

支持C、C++、Python、.NET /Mono、Fortran和Java语言

授权协议： LGPL

开发语言： C/C++

操作系统： 跨平台

官网：http://zeromq.org/,   最新版本4.1.0发布于2014/10/14。Twitter的Storm中使用ZeroMQ作为数据流的传输，还有常见于金融界的应用中。Mongrel2是使用ZeroMQ的一个Web服务器。

 

Jafka/Kafka

 

LinkedIn用Scala语言开发。高吞吐量高性能支持跨语言分布式Publish/Subscribe消息队列系统，而Jafka是在Kafka之上孵化而来的。快速持久化、高吞吐、完全的分布式系统、支持Hadoop数据并行加载。

 

授权协议： Apache

开发语言： Scala

操作系统： 跨平台

 

ActiveMQ

 

居于两者(RabbitMQ & ZeroMQ)之间，类似于ZeroMQ，它可以部署于代理模式和P2P模式。完全支持JMS1.1和J2EE 1.4规范。跨平台的，多种语言和协议编写客户端,Java, C, C++, C#, Ruby, Perl, Python, php。应用协议: OpenWire, Stomp REST, WS Notification, XMPP, AMQP。如需配置ActiveMQ则需要在目标机器上安装Java环境。支持集群，同等网络，自动检测，TCP，SSL，广播，持久化，XA，多个消息也可以组成原子事务

缺点：默认的配置性能偏低，需要优化配置，但是配置文件复杂，本身不提供管理工具；示例代码非常少；主页上的文档看上去比较全面，但是缺乏一种有效的组织方式，文档只有片段，用户很难由浅入深进行了解，二来文档整体的专业性太强。

授权协议： Apache

开发语言： Java

操作系统： 跨平台

 

Apollo

 

ActiveMQ的下一代产品为Apollo，Apollo以ActiveMQ原型为基础，是一个更快、更可靠、更易于维护的消息代理工具。Apache称Apollo为最快、最强健的STOMP（Streaming Text Orientated Message Protocol，流文本定向消息协议）服务器。

l  Apollo的特性如下：

l  支持Stomp 1.0和Stomp 1.1协议

l  主题和队列

l  队列浏览器

l  主题持久订阅

l  镜像队列

l  可靠的消息传递

l  消息过期和交换

l  消息选择器

l  JAAS验证

l  基于ACL的授权

l  支持SSL/TLS，证书验证

l  REST Management API

http://activemq.apache.org/apollo/

 

Redis

一个Key-Value的NoSQL数据库，开发维护很活跃，虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。（盛大开源的DOMQ用的是redis）

 

MemcacheQ

国人开发的持久化消息队列memcacheq(简称mcq)是一个轻量级的消息队列，MemcacheQ的特性：

l  简单易用

l  处理速度快

l  多条队列

l  并发性能好

l  与memcache的协议兼容。这就意味着只要装了memcache的extension就可以了，不需要额外的插件。

l  在zend framework中使用也很方便。

官网：http://memcachedb.org/memcacheq/  最新版本0.2.0发布于2009/12。新浪微博有用。

 

Open-MQ

一个开源的消息中间件，类似IBM的 WebSphere MQ(MQSeries)，采用 C++ 和 Qt 库编写的，支持Windows、Unix 以及 Mac OS 平台，支持 JMS。

授权协议： 未知

开发语言： C/C++

操作系统： 跨平台

官网： http://www.open-mq.com/ (打不开)  目前Open Message Queue 已经集成到了 GlassFish 和OpenESB中。

 

ActiveMQ-CPP

CMS (全称是 C++ Messaging Service) 是一个 C++ 实现的类似 JMS 的 API，用于实现例如 ActiveMQ 的消息代理服务。CMS 可以帮助你的 C++ 客户端代码更见简单。ActiveMQ-CPP 是一个纯客户端库，用它来跟例如 ActiveMQ 等消息服务通讯。我们的 CMS 实现名为 ActiveMQ-CPP，使用可插入式的传输和协议，当前支持 OpenWire 和 Stomp协议，基于 TCP 和 SSL 。同时支持故障转移传输。

 

授权协议： Apache

开发语言： C/C++

操作系统： 跨平台

 

MQ4CPP(Message Queuing for C++)

MQ4CPP is a Message Oriented Middleware (MOM) and implements the following messaging paradigms:

– Direct/Indirect messaging (local)

– Unsolicited messaging (remote)

– Request/Reply (remote)

– Conversation (remote)

– Broadcast (local/remote)

– Publish/Subscribe

– Store & Forward

– Memory Channel

– File Transfer

– Distributed Lock Manager

Support of:

– Multithreading (pthread, WinThread)

– Sockets (berkley , Win Sock2)

– Cluster (failover, session replication)

– Encription (Rijndael 128/256)

– Compression

– Service lookup (local/remote)

– Message routing

Tested platforms:

– Linux (x86, IA64) POSIX

– Windows (x86, IA64) SDK

 

授权协议：LGPL

开发语言：C++

操作系统：跨平台

官网：http://www.sixtyfourbit.org  域名已失效，资料甚少。仅有个overview PDF文档链接：

http://blog.chinaunix.net/attachment/attach/24/54/45/42245445426591fadde801572bf29de4e1f5bd0f37.pdf

最新版本1.16发布于2007年 http://freecode.com/projects/mq4cpp

 

Metamorphosis (MetaQ)

一个高性能、高可用、可扩展的分布式消息中间件，Linkedin开源MQ——Kafka的Java版本，阿里巴巴对此做了定制和优化，具有消息存储顺序写、吞吐量大和支持本地和XA事务等特性，适用于大吞吐量、顺序消息、广播和日志数据传输等场景

 

开发语言： Java  

操作系统： 跨平台

官网： http://metaq.taobao.org/   https://github.com/killme2008/Metamorphosis/

 

Beanstalkd

一个简单、快速的消息队列。Beanstalk之于RabbitMQ，就好比nginx之于Apache，Varnish之于Squid。简单、轻量级、高性能、易使用等特点，以及优先级、多队列、持久化、分布式容错、超时控制等特性。Beanstalkd 包含多种编程语言的客户端开发包。Beanstalkd是典型的类Memcached设计。

不足就是尚无提供删除一个tube的操作，只能将tube的job依次删除，并让Beanstalkd来自动删除空tube。还有就是Beanstalkd不支持客户端认证机制(开发者将应用场景定位在局域网)。没有提供主从同步+故障切换机制，在应用中有可能成为单点的风险。在实际应用中，可以使用数据库为job提供持久化存储。和Memcached类似，Beanstalkd依赖libevent单线程事件分发机制，不能有效的利用多核cpu的性能。这一点可以通过单机部署多个实例克服。

授权协议： MIT

开发语言： C/C++

操作系统： 跨平台

链接：https://github.com/kr/beanstalkd  目前支持过有9.5 million用户的Facebook Causes应用。PostRank大规模部署和使用，每天处理百万级任务

 

OpenAMQ

一个分布式的消息通讯框架。消息异步传递的。主要为高性能和可靠性而设计。该项目服务器端采用 GPL 授权协议，客户端（Python、Java、Ruby、C）是 BSD 授权协议。iMatrix公司早期也是计划按照AMQP/1.0标准开发一个类似RabbitMQ的项目，名字叫做OpenAMQ。然而，随着项目的推进，iMatrix公司在这个项目上迷失了方向。iMatrix的CEO Pieter Hintjens在一篇文章中描述了自己对AMQP标准化进程的困惑和思考，并认为AMQP中存在一些无法克服的问题。2010年3月，iMatrix公司宣布退出AMQP/1.0标准化，放弃OpenAMQ项目，并正式启动了ØMQ，即ZeroMQ

授权协议： GPLv2/BSD

开发语言： C/C++

操作系统： 跨平台

官网：http://www.openamq.org/  目前最新版1.4发布于2010/10/7

 

The Spread Toolkit

 

高性能的分布式分组消息系统，支持局域网以及广域网通讯. Spread可以作为一个分布式应用的消息总线，并且具有高度的灵活性，可以做到多播，分组，以及点对点饿消息传递。

The Spread toolkit 包括一个消息服务器 server，以及多种语言的api C/C++ libraries (with and without thread support), a Java Perl, Python, and Ruby. 还有很多其他语言的第三方扩展。

在一个典型的环境中，通常每台服务器上运行一个Spread server，客户端的程序本地连接server，发送信息，而这台服务器上的spread server会传递信息给其他订阅了这条消息的应用。当然也可以只有一个spread server，而其他的客户端分布在整个网络中。

Some of the services and benefits provided by Spread:

l  Reliable and scalable messaging and group communication.

l  A very powerful but simple API simplifies the construction of distributed architectures.

l  Easy to use, deploy and maintain.

l  Highly scalable from one local area network to complex wide area networks.

l  Supports thousands of groups with different sets of members.

l  Enables message reliability in the presence of machine failures, process crashes and recoveries, and network partitions and merges.

l  Provides a range of reliability, ordering and stability guarantees for messages.

l  Emphasis on robustness and high performance.

l  Completely distributed algorithms with no central point of failure.

 

授权协议： Spread Open-Source License(类似BSD)&&商业协议

开发语言： C/C++

操作系统： 跨平台

官网： http://www.spread.org/ 目前最新版本4.4.0发布于2014/5/28，目前所知国内豆瓣在使用。国外有研究项目Secure Spread、Wackamole、Oasis、MEAD、ATLAS RepDB*等，国外应用有mod_log_spread、Fedora Unity Spread Logging、Splash、MySQL Message API、Zope Replication Services、Spread Management Tools、Ssrc Wisp™ Event-based distributed service framework、Perl Messaging:Courier library、Perl Spread-Queue module等等。

 

SAFMQ (Store and Forward Message Queue)

消息队列服务器提供了异步的、round-trip、可靠的消息传输。

一个简单的消息中间件，采用C++编写，截至2006年11月SAFMQ的的版本为0.5.2，具有如下的功能： 1. 提供多队列、多优先级的消息转发服务。 2. 支持文本、二进制的消息类型。 3. 支持转发功能，即多个消息中间件之间的消息转发。 4. 支持事务操作 5. 支持Java、PHP客户端 6. 支持SSL加密 7. 支持用户权限 8. 支持对消息的标记 9. 支持TTL(Time To Live)时间戳。目前最新版本0.8.3.1，2010年

授权协议： Apache

开发语言： Java C/C++ PHP

操作系统： 跨平台

官网： http://safmq.sourceforge.net/  ，最新版本0.8.3.1发布于2010/11/3

 

Mosquitto

一个开源(BSD许可证)的消息代理，实现MQTT(消息队列遥测传输)协议版本3.1。提供了Windows、Linux以及qnx系统的版本。MQTT是轻量级基于代理的发布/订阅的消息传输协议。 MQTT是IBM开发的一个即时通讯协议。MQTT是面向M2M和物联网的连接协议，采用轻量级发布和订阅消息传输机制

有三种消息发布服务质量：

“至多一次”，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。

“至少一次”，确保消息到达，但消息重复可能会发生。

“只有一次”，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果。

授权协议： BSD

开发语言： C/C++

操作系统： 跨平台

官网： http://mosquitto.org/ 最新版本1.3.5发布于2014/10/8

 

MUSCLE (Multi User Server Client Linking Environment)

提供一个多路的消息服务器以及相应的网络API，其客户端涉及多种语言包括 C, C++, C#, Delphi, Java, 和 Python。MUSCLE 用来在网络上传输消息对象，所有消息存储在服务端并为客户端进行传递。

授权协议： BSD

开发语言： Java C/C++ Python C#

操作系统： 跨平台

官网：https://public.msli.com/lcs/muscle/ 目前新版6.10发布于2014/12/12。MUSCLE has been developed, used, and refined as the networking component of BeShare, CueStation and various other audio control applications at Meyer Sound Laboratories for over twelve years

 

JORAM

JORAM一个类似于openJMS分布在ObjectWeb之下的JMS消息中间件。文档非常完备，并且带有很多示例。

缺点：

判断JMS客户端是否在线非常缓慢，有时甚至不会通知应用

授权协议： 未知

开发语言： Java

操作系统： 跨平台

官网：http://joram.objectweb.org/

 

 

三、     MQ比较

一些国外网站提供的，ActiveMQ、RabbitMQ、RocketMQ（MeteMQ）、HornetQ、Qpid、ZeroMQ的对比数据。

协议支持比较：

 

ActiveMQ

Apollo

HornetQ

Qpid

RabbitMQ

ZeroMQ

AMQP

1.0

1.0

announced

1.0

0-8, 0-9, 0-9-1

-

MQTT

    

-

-

  

-

OpenWire

    

-

-

-

-

REST

      

-

  

-

STOMP

      

-

  

-

STOMP over Websockets

      

-

  

-

XMPP

  

-

-

-

Over Gateway

-

 

客户端接口支持比较：

 

ActiveMQ

Apollo

HornetQ

Qpid

RabbitMQ

ZeroQ

C

  

-

-

      

C++

-

-

-

      

Erlang

-

-

-

-

    

Haskell

-

-

-

-

    

Java JMS

  

-

    

-

-

Java proprietary

  

-

  

-

    

.NET

-

-

-

      

Objective-C

-

-

-

-

-

  

Perl

-

-

-

-

    

PHP

-

-

-

-

    

Python

-

-

-

      

Ruby

-

-

-

      

 

性能测试场景对比：

l  情景A：先入队20,000 条1024 字节大小的消息, 然后再出队

l  情景B：20,000条1024字节大小的消息同时入队和出队

l  情景C：200,000 条32 字节大小的消息同时入队和出队

l  情景D：200 条32K字节大小的消息同时入队和出队

 

两种不同配置的broker，一种开启持久化消息(Persistent)，一种是没有开启持久化，即瞬时化消息(Transient)

下面是测试的所有brokers和对应的配置:

 

 

情景 A：

 

情景 B：

 

 

情景 C：

 

情景 D：

 

结论：

1)        Brokers普遍擅长于处理大消息。因此如果客户端支持对消息分组，那性能会得到更大的提升。但分组消息却不能在consumer之间传播。

2)        处理大消息，持久化的弊端（磁盘或数据库保存）就开始凸显 (QPID 处理瞬时化消息不论消息大小，都显得非常高效)。可以得出，处理中小消息的耗时主要在集中CPU，而不是I/O网络上。

3)        ZeroMQ broker比其他所有MQ broders表现得更优越。倘若需求上要用到一些特殊的broker特性，不然ZeroMQ 绝对是分发消息系统的一个最好选择。

4)        QPID 似乎是在处理瞬时消息上综合表现得最好。

5)        从RabbitMQ的测试结果看，AMQP 协议似乎比STOMP协议更优越. 当然这结果也可能跟不良设计的客户端受到影响。

6)        HornetQ 在处理中、小消息时，表现得最差的。

7)        除了处理大消息之外，RabbitMQ 似乎是最好的选择，因它的性能是其他的3倍左右。

下图是显示的是发送和接受的每秒钟的消息数。整个过程共产生1百万条1K的消息。测试的执行是在一个Windows Vista上进行的。

 

 

选择消息系统根据业务需要需要考虑以下几个方面：

l  是否持久化

l  吞吐能力

l  高可用，避免单点故障

l  分布式扩展能力

l  兼容现有协议

l  易于维护

l  其他，如消息丢失和重复的处理

l  负载均衡

 

常见消息系统协议：

l  STOMP

l  AMQP

l  类似 MEMCACHE 的协议

l  HTTP

l  MQTT

 

非C/C++实现服务端的，资料文档缺乏的，版本最后更新时间久远的，活跃度低的，重量级的，这些MQ排除不考虑。下面是筛选出来的若干个MQ作对比

 

Qpid

ZeroMQ

Beanstalkd

Spread

Open-MQ

memcachedQ

SAFMQ

Mosquitto

MUSCLE

JMS

━

━

━

━

√

━

━

━

━

AMQP

√

━

━

━

━

━

━

━

━

MQTT

━

━

━

━

━

━

━

√

━

OpenWire

━

━

━

━

━

━

━

━

━

Stomp

━

━

━

━

√

━

━

━

━

Memcache协议

━

━

√

━

━

√

━

━

━

HA集群(防单点故障)

√

√

━

√

━

━

━

━

━

认证

√

━

━

√

━

━

√

━

━

Broker架构

√

━

√

√

√

√

√

√

√

持久化

√

━

√

√

━

━

√

━

━

支持广域网

━

━

━

√

━

━

━

━

━

高吞吐

√

√

√

√

√

√

√

━

━

事务

√

━

━

━

━

━

━

━

━

 

下面对非持久化消息进行了测试(ZeroMQ自实现一个简单broker，直接内存操作和转发)

测试硬件环境：

Broker:

Operation System:         Windows7 旗舰版  sp1  x64

CPU:                    Intel® Core™ i5-3470 CPU @ 3.20GHz

MEM：                4.00GB

Disk:                     500GB

Network Adapter:          Gibabit  Network connection

 

Client:

Operation System:         Windows7 旗舰版  sp1  x64

CPU:                    Intel® Core™ 2 Duo CPU T6400 @ 2.00GHz

MEM：                4.00GB

Disk:                     250GB

Network Adapter:          Gibabit  Network connection

 

测试结果：

 


 

 

结论：

1)      瞬时化消息的处理性能明显大大优于持久化消息的处理

2)      两者都更擅长处理大消息体数据

3)      处理的消息体越小时，Qpid的性能下降得比较明显

4)      Qpid在处理持久化消息时，消息体越大，性能越高。这说明消息体比较大的情形，瓶颈在于网络IO，消息体越小，瓶颈在于CPU和磁盘读写。

 

四、     ZeroMQ简介

ZeroMQ用于node与node间的通信，node可以是主机或者是进程。ZeroMQ 把通讯的需求看成四类。其中一类(Exclusive-Pair)是一对一对应通讯，用来支持传统的 TCP socket 模型，但并不推荐使用。常用的通讯模式只有三类。

  1. 请求回应模型（Request-Reply）。

由请求端发起请求，并等待回应端回应请求(阻塞的)。请求端和回应端都可以是1:N的模型。通常把1认为是server，N认为是Client。ZeroMQ 可以很好的支持路由功能（实现路由功能的组件叫作 Device），把 1:N 扩展为 N:M。

            

  1. 发布订阅模型（Publish-Subscribe）。

发布端是单向只发送数据的，且不关心是否把全部的信息都发送给订阅端。如果发布端开始发布信息的时候，订阅端尚未连接上来，这些信息直接丢弃。不过一旦订阅端连接上来，中间会保证没有信息丢失(之前的消息会丢掉，Slow joiner问题)。同样，订阅端则只负责接收，而不能反馈。Publisher 中途离开，所有的 Subscriber 会 hold 住，等待 Publisher 再上线的时候，会继续接受信息。如果发布端和订阅端需要交互（比如要确认订阅者是否已经连接上），则使用额外的 socket 采用请求回应模型满足这个需求。

           

  1. 管道模型（Push-Pull）。

这个模型里，管道是单向的，从 PUSH 端单向的向 PULL 端单向的推送数据流。

 

任何分布式并行的需求，都可以用这三种模型组合起来解决问题。ZeroMQ 只专注和解决了消息通讯这一基本问题。

ZeroMQ 中的 Transient (短暂) 和 Durable (持久) socket 也并非区别于实现层是否保持了 tcp 连接。而是概念上的不同。对于 Durable socket ，其生命期可以长于一个进程的生命期，即使进程退出，再次启动后依旧可以维持继续之前的 socket。

l  zmq_init创建一个context，可以认为是一个MQ实例或句柄。1表示IO线程数。

l  zmq_socket根据context来创建一个socket,后面类型指定了MQ通信类型。

l  zmq_bind/zmq_connect可以进行绑定进行监听或者是进行连接。

l  zmq_msg_init/zmq_msg_init_size可以用来初始化一个message

l  zmq_send/zmq_recv可以进行message的发送和接收。

l  zmq_msg_close销毁一个message

l  zmq_close关闭一个socket

l  zmq_term销毁一个context

 

通信协议：

l  tcp // 跨主机间通信

l  ipc // 进程间通信

l  inproc // 线程间通信

l  pgm // ━━━

l  epgm // ━━━

 

消息：

ZeroMQ通信通信单元是消息，他除了知道 Bytes 的大小，他并不关心的消息格式。zmq_recv/zmq_send只能够处理内置的消息格式，而不能够处理http请求这种字节流。ZMQ允许一条message按照多个部分进行发送(multipart message)。底层使用其他线程完成了IO读写。ZMQ内置有一个字节流成帧策略。

 

Identity：可以用来表示一个socket的身份

Device：一旦通信节点超过一定数量的话，那么最好需要一个转发节点或者是中间节点。

 

拥塞：

ZMQ可以通过控制HWM(high-water mark)来控制拥塞。内部实现上每一个socket有关联了buffer,HWM可以控制buffer大小

l  PUB/PUSH有transmit buffers.

l  SUB/PULL/REQ/REP有receive buffers.

l  DEALER/ROUTER/PAIR有transmit buffers也有receive buffers.

一旦socket达到了high-water mark的话，那么会根据socket类型来决定是丢弃还是block.现在实现而言的话PUB会尝试丢弃数据，而其他类型的socket就会block住。 如果socket是线程之间进行通信的话，那么HWM是两者socket的HWM之和。因为默认HWM是ulimited的，所以只要一端没有设置的话那么容量就无限。

 

官网：http://zeromq.org/, 最新版本4.1.0发布于2014/10/14。Twitter的Storm中使用ZeroMQ作为数据流的传输，还有常见于金融界的应用中。

关于ZeroMQ的权威资料，除了官方文档，还有O\'Reilly出版社2013年出版的一本书，《ZeroMQ: Messaging for Many Applications》，作者: Pieter Hintjens

 

目前最新稳定版本4.0.5源代码中，C/C++代码一共大概1.7万行，包括主体代码占1.3万左右，API和Demo占4千行左右。详细如下图。

 

 

 

 

五、     Qpid简介

基本 Qpid 通信系统的几个组件

 

Address 地址

Qpid Address 表示一个节点，有两种节点：一种是 queue，另外一种是 topic。queue映射到 AMQP 概念就是 Queue；而 topic则映射到 Exchange。Queue 节点能够缓存消息，直到被读取走为止；而 topic 节点则即时进行转发，比如假如有 4 个 consumer 对某消息感兴趣，当消息到达节点时，有 3 个 consumer 正在运行，那么 topic 节点会将消息转发给这 3 个 consumer，然后就将该消息丢弃。剩下的那个 consumer 再运行时，则收不到这个消息。

Address 是一个带格式的字符串，其语法如下：

 address_string ::= <address> [ / <subject> ] [ ; <options> ]

 options ::= { <key> : <value>, ... }

 

其中 address，subject 和 key 都是字符串。

Subject 类似 email 的主题。每个消息都可以有一个主题，接收者可以通过主题对消息进行过滤。

Option 的具体含义有点儿复杂，可以参考 Qpid 的编程手册获取完整的描述。

 

Broker Federation

单一 Broker 的构架

 

Broker Federation 的配置

 

设置 federation 的命令叫 Qpid-route。Qpid 支持两类路由：Queue routes 和 Exchange Routes。

Queue route：目的地必须是 Exchange，源是一个 Queue。此类路由将源地址 Queue 收到的所有消息都转发到目的 Exchange 去。

Exchange Route：目的地依然必须是一个 Exchange，源也是一个 Exchange。此类路由允许将源 Exchange 上收到的，拥有指定 RouteKey 的消息转发到目的 Exchange 上去。

 

RDMA

RDMA 全称为“Remote Direct Memory Access”，它是一种协议，将数据从一台计算机的内存直接传输到另外一台计算机的内存中，而无需 CPU 的参与。类似 CPU 与外设之间的 DMA 传输，RDMA 将 DMA 的概念扩展到了网络。

C++ 版本的 Qpid broker 除了使用传统的 TCP/IP 作为网络通信机制之外，在拥有 infiniband 设备的集群上 Qpid 还可以使用 RDMA 进行网络通信。

 

持久化消息

Broker 将收到的消息暂存在磁盘等可以永久保存信息的地方备用，这样，即使 Broker 因为某种意外而中断，当其再次重新启动之后，还是可以将保持在永久存储介质中的消息读出来并继续进行转发。除了 Queue 需要 durable 之外，我们还必须保证发送的消息有 durable 属性。

 

高可用性

在一个 Qpid broker 集群中，所有的 broker 都互相备份，进行 fail over 的必要准备工作。每个 broker 的内部对象都同步到其他集群中的 Broker，保持一致，这样在单一 Broker 无法工作的情况下，client 可以切换到其他 Broker，而避免信息的丢失和服务中断。

Qpid 和 Corosync 的工作模式

Qpidd A 和 QpiddB 通过 Corsync 同步。Qpid 还通过 CMAN 来防止集群中的”split brain”问题，CMAN 提供了 quorum 算法，Qpidd 利用 CMAN 的接口，知道自己是否能够达到法定人数，是否能够加入集群工作

Qpid 集群是一个 Active/Active 模式的集群。客户端可以使用任意一个 broker。如下左图所示：

 

当 client 连接到一个集群中的 broker 时，该 broker 返回给 Client 相应的 Broker URL 列表。在上图中，Client 将得到 [QpiddA,QpiddB] 这样一个列表。当 QpiddA 的连接断开时，客户端可以自动重新连接到 QpiddB 继续服务。

 

最新稳定版本0.30发布于2014-09-26。2015-01-27发布修复几个漏洞的0.31版本补丁。0.30版本源代码中，C/C++代码一共大概10万行，其中主体代码占大概8万，客户端API及Demo占大概2万。官方0.30版本整个源码包统计详细结果如下图。

 

 

参考文献：

1、http://predic8.com/activemq-hornetq-rabbitmq-apollo-qpid-comparison.htm

2、http://www.tuicool.com/articles/JvQRbm

3、http://blog.x-aeon.com/2013/04/10/a-quick-message-queue-benchmark-activemq-rabbitmq-hornetq-qpid-apollo/

4、http://blog.jobbole.com/45547/

5、http://blog.fity.cn/post/377/

6、http://www.ibm.com/developerworks/cn/opensource/os-cn-qpid1/#ibm-pcon

7、http://news.cnblogs.com/n/154000/

 

以上是关于消息中间件的主要内容，如果未能解决你的问题，请参考以下文章

消息中间件

消息中间件专题-消息中间件引言

消息中间件之Kafka

厉害！一文了解消息中间件-RabbitMQ

消息中间件剖析

消息中间件