一种提高消息传递性能的集群通信系统及其方法

摘要

本发明涉及网络通信技术领域，特别涉及一种提高消息传递性能的集群通信系统及其方法，所述系统包括消息发布端、包含多台消息服务器的消息服务器集群、消息订阅端和分布式协调服务集群；本发明具有全局的无单点集群化可扩展的分布式设计；如果消息发布端或订阅端一个或多个节点发生故障，可以由同一组中的其他节点继续发送或接收消息，并不会中断消息的处理流程；采用基于索引的分布式文件存储方案，有效规避了现有的DB和文件存储的缺点，使得消息的读写效率更高；使用长轮询PULL的消息投递方式，在保证消息的实时性的同时兼顾了吞吐量。

说明书

技术领域

本发明涉及网络通信技术领域，特别涉及一种提高消息传递性能的集群通信系统及其方法。

背景技术

在现代分布式系统中，集群中多个节点之间的异步消息传输通常依靠消息系统进行。与原始的点对点通信不同，消息系统在整个应用系统中承担着数据路由的职责，可以有效地对各个子系统进行解耦。

遵循发布订阅模型的消息系统一般由三个对象构成：消息发布者(Producer)、消息服务器(Broker)、消息订阅者(Comsumer)。消息发布者负责产生消息，并将消息发送到消息服务器，消息可以按照主题区分为不同的类别。消息订阅者向消息服务器订阅一个或多个感兴趣的消息类别(Topic)，并且只接收感兴趣的消息。消息服务器则负责存储和转发消息。所述消息系统要将消息发布端发布的消息异步地发送给消息订阅端。

现阶段市面上主要的开源消息中间件产品有Kafka，RabbitMQ，ActiveMQ等，目前这些主流的中间件在可扩展性、持久化、消息的高性能投递方面具有比较明显的缺陷，包括：

在可扩展性方面，现有技术只能保证在消息服务器端的可扩展，不能完全保证消息发布端、消息订阅端这两个点可扩展，处理能力有限，不能完全防止单点问题，例如消息订阅端出现单点故障，就无法从消息服务器获取订阅的消息进而消费消息，从而影响与之关联的其他系统的处理。

在消息的持久化方面，现有产品一般采用数据库(Database，简称DB)方案或文件存储方案。对于采用DB存储方案，则会使用树数据结构B+树作为消息索引，B+树涉及磁盘的随机读写，当消息出现海量堆积时，B+树膨胀造成读写性能急剧下降。而文件存储方案也会频繁地进行磁盘IO读写，成为性能瓶颈。

在消息的高性能投递方面，现有消息系统有推(PUSH)和拉(PULL)两种消息投递模式。PUSH模式是消息服务器主动地把消息推送给消息订阅者，该方式实时性比较高，但对服务器的压力比较大。PULL模式是指客户端主动地向服务器端拉取数据，该方式吞吐量大，但实时性低。这两种投递模型都不能满足同时对实时性和吞吐量都有严格要求的应用场景。

随着云计算和互联网规模的不断扩大，具有高并发和海量消息流转需求的业务场景越来越多，如果沿用传统的消息系统，当面对爆发式增加的访问压力时，传统的消息系统可能会产生消息处理缓慢、消息丢失甚至消息服务器宕机的现象。

发明内容

针对以上技术问题，本发明提供了一种提高消息传递性能的集群通信系统及其方法，采用完全的分布式设计以解决现有技术中的单点问题提高了可扩展能力。同时，为了实现消息投递的高性能，在消息存储、IO、消息负载均衡策略和消息推拉模式等方面进行了优化。

本发明一种提高消息传递性能的集群通信系统，包括消息发布端、包含多台消息服务器的消息服务器集群、消息订阅端和分布式协调服务集群；

所述消息发布端和消息订阅端通过消息服务集群连接，并通过消息服务集群传递消息，消息发布端、消息服务器集群和消息订阅端都与分布式协调服务集群保持长连接；

所述消息发布端按照发布消息的主题Topic类别不同划分为不同的组，用一个groupID作为该组的唯一标识；

所述消息订阅端按照定阅消息的主题Topic类别不同划分为不同的组，用一个groupID作为该组的唯一标识；

所述消息发布端和消息订阅端定时从分布式协调服务集群拉取Topic路由信息并更新到本地，以获取该向哪一台消息服务器发布消息或拉取消息，而每台消息服务器则定时向分布式协调服务集群发布提供存储和转发服务的Topic和IP地址端口信息。

优选地，消息服务器集群将接收的消息按照主题分片保存在不同的消息服务器上。

优选地，为每个存储分片消息的消息服务器增加复制集群，复制集群中的每个节点都存储主节点的同一份数据，复制因子R表示一份数据复制保存在R个不同节点上。

优选地，所述复制集群包括一个主响应机leader和该主响应机leader的至少一个备用响应机follower，最初的主响应机leader通过用户配置确定，当leader失效的时候，由该leader的所有follower投票选举其中之一follower成为新的leader，接替之前失效的leader。

优选地，所述消息服务器集群将接收的消息按照主题分片保存在不同的消息服务器上，包括消息依据不同的Topic保存在不同的逻辑队列中，逻辑队列用来指定消息在真实的物理文件中的偏移位置，指向消息在物理文件中的索引。

优选地，所述物理文件由多个文件SegmentFile构成，SegmentFile则是由若干个不定长的存储单元组成的大小为1GB的文件，每个存储单元指定了这条消息的长度和具体内容。

优选地，消息服务器集群提供的消息服务中所有的消息都是持久化的，即消息的存储和转发利用操作系统提供的页缓存PageCache，如果在PageCache中没有命中数据则再去访问磁盘。

优选地，消息发送端，消息服务器集群，消息订阅端两两之间底层数据通信采用推拉结合的长轮询的消息投递机制，消息服务器集群中某个节点根据实际消息的更新情况来处理消息订阅端发送过来的消息拉取请求，即如果无最新消息，服务器就会将请求阻塞，直到有新消息需要传递或超时时才返回；消息订阅端收到服务器发送回来的消息或者控制信息后调用处理函数处理完信息后，再次发送请求消息的长连接请求，继而等待消息到达，进入下一个循环。

本发明一种提高消息传递性能的集群通信方法，包括：

消息发布端初始化要发送消息并指定其Topic；消息发布端将本地的Topic路由信息定时与协调子系统同步，然后通过Topic路由信息确定该消息该发往哪一个消息服务器；消息服务器收到消息后，持久化到其文件系统，即首先写入PageCache，当写满一定的页数时再批量flush到磁盘；消息订阅端订阅Topic；消息订阅者向消息服务器拉取消息。

优选地，消息订阅者拉取消息时进行负载均衡，即每个订阅者消费一个Topic下的个逻辑队列，消费完成后删除消息服务器上存储的此条消息；N为该Topic下的逻辑队列数量，M为订阅组中的订阅者数量，表示向下取整运算。

本发明与现有方案相比，具有以下有益效果：

全局的无单点集群化可扩展的分布式设计。如果消息发布端或订阅端一个或多个节点发生故障，可以由同一组中的其他节点继续发送或接收消息，并不会中断消息的处理流程。采用基于索引的分布式文件存储方案，有效规避了现有的DB和文件存储的缺点，使得消息的读写效率更高。使用长轮询PULL的消息投递方式，在保证消息的实时性的同时兼顾了吞吐量。

附图说明

图1是本发明提高消息传递性能的集群通信系统优选实施例结构示意图；

图2是本发明消息服务器内部结构图；

图3是本发明消息通过异步复制线程在不同的存储节点中保存示意图；

图4是本发明基于索引的消息存储数据结构；

图5是本发明基于长轮询的消息投递模型；

图6是本发明提高消息传递性能的集群通信方法第一优选实施例的流程图；

图7是在高并发连接情况下本发明与现有系统消息时延性能对比示意图；

图8是在高并发连接情况下本发明与现有系统每秒成功处理消息数量对比示意图。

具体实施方式

为了使本发明的目的、技术方案及优点更加清楚明白，以下结合附图对本发明实施例进一步详细说明。

本发明系统及方法实施例基于相同发明构思，为节省篇幅，两者之间相同的技术描述分别有所侧重，但两者相应的技术描述可以相互引用。

本发明实施例设计了一种提高消息传递性能的集群通信系统，是对现有技术中分布式消息中间件的改进，如图1所示，是所述系统优选施例整体架构示意图。

所述系统包括消息发布端、消息服务器集群、消息订阅端和分布式协调服务集群。

所述消息发布端和消息订阅端通过消息服务集群连接，并通过消息服务集群传递消息，同时消息发布端、消息服务器集群和消息订阅端都与分布式协调服务集群保持长连接，也即持久连接，链路保持TCP连接不断开。

与目前开源的主流消息中间件不同，本发明所述消息发布端和消息订阅端不是单个独立的节点，而是按照发布和订阅消息的主题Topic类别不同划分为不同的组(例如，消息发布组1、消息发布组2)，并且用一个groupID作为该组的唯一标识。另外，消息发布端和消息订阅端定时从分布式协调服务集群拉取Topic路由信息并更新到本地，以获取该向哪一台消息服务器发布消息或拉取消息，而每台消息服务器则定时向分布式协调服务集群发布提供存储和转发服务的Topic和IP地址端口信息。

Topic路由是由很多键值对构成的集合(一个键值对是指Topic和存储此Topic的消息服务器节点地址的对应关系)，键就是Topic本身的内容，值是负责存储此Topic的消息服务器的IP地址(可能有多个)。当发布端发送消息时会根据Topic到路由信息中查询此消息应该发到哪个消息服务器(通过下述的负载均衡策略选择一个消息服务器发送)

消息服务器集群中包含多台消息服务器，每台消息服务器的消息存储架构类似于MongDB的架构，将接收的消息按照主题分片保存在不同的消息服务器上。

同时为了保证消息的高可用，防止单点问题发生，为每个存储分片消息的消息服务器增加复制集群(复制集群采用了冗余存储的方法保证数据安全，复制集群中的每个节点都存储主节点的同一份数据，防止单点故障导致的数据丢失)，复制因子R表示一份数据复制保存在R个不同节点上，R的取值一般由系统服务的可用级别确定，R＝2时提供简单的冗余和服务器故障保护，当R＝3时则能保证在系统灾难性故障情况下数据不丢失。

复制集群指的是：使用多台主机(一般为2台)组成一个集群，这个集群中的每台主机负责冗余存储某个消息服务器(其实就是下述leader)的所有消息，防止这个消息服务器发生单点故障从而导致数据丢失。下面的follower其实就是属于这个复制集群中的一台主机(或一个节点)。

基于复制集群的方案，那么就意味着需要对多个备份进行调度，每个分片都有一个消息服务器作为主响应机leader，leader负责所有的读写操作，如果leader失效，那么将会有其他备用响应机follower来接管(成为新的leader)。follower只是单纯地与leader同步消息即可。由此可见作为leader的server承载了全部的请求压力，因此从集群的整体考虑，有多少个分片就意味着有多少个leader，系统会将leader均衡地分散在每个分区上来确保整体的性能稳定。

最初的leader是通过用户配置确定的，当leader失效的时候，会通过raft算法，由该leader的所有follower投票选举(随机投给自己或其他follower)，一旦某个follower的票数超过半数(如果投票结果没有超过半数的会继续重新投票直到出现为止)，那么这个follower就会成为新的leader，接替之前失效的leader继续向外提供服务。

在图2中，消息发布组1发布了m11、m12、m13、m14，4个消息，其中m11、m12、m14属于TopicA，ma13属于TopicB；消息发布组2发布了m21，m22，m23，3个消息，其中m22属于TopicA，m21和m23属于TopicB。消息依据不同的Topic保存在不同的逻辑队列中，逻辑队列则相当于字典目录用来指定消息在真实的物理文件中的偏移位置，同时，如图3所示，消息会通过异步复制线程在不同的存储节点中保存R份。业界一般认为当一份数据保存3份即可以保证数据99％不丢失，所以一般只需要复制两份即可。

为了减少系统频繁读取大文件对磁盘IO和内存造成巨大压力，如图4所示，本文采用了一种索引存储的数据结构将大文件拆分成小文件来提高持久化性能。

消息按主题进行划分存储，每个主题下面都有多个队列TopicQueue，这个队列其实是一个逻辑队列(一种按照先进先出顺序存储消息的数据结构，逻辑队列存储的不是消息本身，而是消息在Linux文件中的具体位置，相当于一个索引，)，指向消息在物理文件中的索引。真正存储消息的物理存储结构SegmentLsit是由多个文件SegmentFile构成，SegmentFile则是由若干个不定长的存储单元组成的大小为1GB的文件，每个存储单元指定了这条消息的长度和具体内容。

优选地，消息服务器集群提供的消息服务中所有的消息都是持久化的，为了尽量减少耗时IO操作，充分提高系统性能，消息的存储和转发可以利用操作系统提供的页缓存PageCache，如果在PageCache中没有命中数据则再去访问磁盘。所述消息服务是指整个消息系统对其他分布式应用提供的消息存储和转发的服务。所谓持久化是指将消息存储在磁盘这类外部存储器上，而不是保存在内存上，防止断电造成存储的内容消失，从而实现“持久存储”。

消息的刷盘策略(消息从内存中写入到磁盘中的方式)分为同步刷盘和异步刷盘。同步刷盘是指Producer(消息发布端中的某个节点)发送消息到Broker(消息服务器集群中的某个节点)保证消息持久化到磁盘再返回。异步刷盘是指Producer发送消息到Broker后立马返回，由后台线程执行异步刷盘操作，每刷满一定页数的PageCache消息才会持久化，即写入到磁盘。

在Broker中使用了零拷贝技术，mmap调用将消息copy到内核缓冲区，此时mmap调用返回消息在内核缓冲区的起始位置给应用程序，应用程序再通过write系统调用，将消息copy到socketBuffer，最后通过网卡发送出去，这样做的好处是避免了数据在内核缓冲区和用户缓冲区之间的相互copy，提高了消息接收和发送的效率。

所述mmap调用是Linux下的系统调用，是一种内存映射文件的方法，即将一个文件或者其它对象映射到进程的地址空间，实现文件磁盘地址和进程虚拟地址空间中一段虚拟地址的一一对映关系，通过这种方法可以有效提高IO效率。上述write系统调用也是linux的一个函数。

网络IO也是消息投递性能和吞吐量的主要瓶颈之一，本文在提高系统网络IO性能方面主要做了两方面的努力，使用了高性能的异步IO框架和Linux的零拷贝技术。

本文在设计网络通信层(是对消息发送端，消息服务器集群，消息订阅端两两之间底层数据通信接口的封装)时，使用了Java NIO框架Netty，相对于传统的同步阻塞式IO，NIO采用了Reactor模式，可以大大提高服务器端的并发连接量，同时NIO是异步的，也提高了数据的传输效率。

在消息推送模型的设计上，针对推、拉模式的优缺点互补的特性，并结合异步JavaSript和XML(Asynchronous Javascript And XM，简称Ajax)长连接模型，本文提出了一种推拉结合的长轮询的消息投递机制，可用于消息发送端，消息服务器集群，消息订阅端两两之间底层数据通信

如图5所示，所述推拉结合长轮询的消息投递机制具体的实现过程为：消息服务器集群中某个节点(消息服务器)根据实际消息的更新情况来处理消息订阅端发送过来的消息拉取请求，即如果无最新消息，服务器就会将请求阻塞，直到有新消息需要传递或超时时才返回。消息订阅端收到服务器发送回来的消息或者控制信息后调用处理函数处理完信息后，再次发送请求消息的长连接请求，继而等待消息到达，进入下一个循环。消息服务器总是有源源不断的消息到达，如果这时候消息订阅端正在处理之前接收到的消息或者刚发送完请求还没有建立连接，在这种连接暂时间断的情况下，服务器会采取一定的保护措施，一般是将这些刚到的消息保存在本地，等到连接再次建立后，服务器会把所有保存的消息和最近更新的消息一次性推送到订阅端。

图6是本发明提高消息传递性能的集群通信的方法第一优选实施例的流程图。如图所示，该消息传递方法中主要步骤包括：

1.首先消息发布端初始化要发送消息并指定其Topic。

2.消息发布端将本地的Topic路由信息定时与协调子系统同步，然后通过Topic路由信息确定该消息该发往哪一个消息服务器，这样就实现了发送方的负载均衡。

3.消息服务器收到消息后，持久化到其文件系统，即首先写入PageCache，当写满一定的页数时再批量flush到磁盘

4.消息订阅端订阅Topic，特别说明的是，此步骤与步骤1没有先后关系，见下述实施例所述。

5.消息订阅者向消息服务器拉取消息，

优选地，消息订阅者拉取消息时，进行负载均衡。前面介绍过，消息订阅者可以是一个组，为了保证这个消息订阅组里的每一个订阅者都能够平均地消费消息，本文采用了类似于操作系统分页的算法。同一个Topic下有N个逻辑队列，假如订阅组中有M个订阅者，那么每个订阅者会消费这个Topic下的个逻辑队列。消费完成后删除消息服务器上存储的此条消息。表示向下取整运算。

本发明提高消息传递性能的集群通信的方法第二优选实施例，具体步骤包括：

1.消息订阅端首先向分布式协调子系统发送订阅请求，分布式协调子系统负责维护整个消息系统的路由信息，根据订阅请求建立起Topic和订阅端之间的映射关系。

2.消息发送端初始化消息并设置消息Topic信息，然后向消息服务器集群发送此消息。

为了实现发布消息的负载均衡，消息发布端会与分布式协调子系统保持心跳，即消息发布端与分布式协调子系统定时进行数据交互，并定时从分布式协调子系统获取消息服务器集群中每个节点的地址路由信息，更新到本地内存，消息发布端发送消息时会轮询地去选择本次消息该发送到哪个消息服务器节点。

3.消息服务器接收到消息以后，首先写入PageCache，当写满一定的页数时再批量flush到磁盘。

消息持久化到磁盘又分为两个具体步骤，一是消息首先写入物理文件并返回该消息的在物理文件中的实际偏移地址，二是再把消息的实际偏移地址按照FIFO的顺序放入消息的逻辑队列，逻辑队列存储的其实是消息在物理文件中的索引。这种索引存储的数据结构将大文件拆分成小文件来提高了持久化性能。另外为了保证高可用，消息服务器采用master-slave的架构，每一个消息服务器会把消息数据同步到其他节点上，以防止出现单点故障造成消息丢失。

4.消息订阅端拉取消息时，需要进行负载均衡。

消息订阅者可以是一个组，这个消息订阅组里的每一个订阅者都能够平均地消费消息。具体类似于操作系统分页的算法。同一个Topic下有N个逻辑队列，假如订阅组中有M个订阅者，那么每个订阅者会消费这个Topic下的N/M个逻辑队列。在确定拉取的目标服务器后消息订阅端会采用长轮询PULL的方式去拉消息。长轮询PULL类似于Ajax的长轮询，它综合了PUSH和PULL模型的优点，消息服务器会根据实际消息的更新情况来处理消息订阅端发送过来的消息拉取请求，如果无最新消息，服务器就会将请求阻塞，直到有新消息需要传递或超时时才返回。这种方法即保证了实时性同时又兼顾了吞吐量。

5.消息订阅端收到消息后，根据自己的消息消费逻辑消费消息，消费完成后向消息服务器发送ACK，随后消息服务器会从磁盘中删除本条消息。

以下将对本发明技术方案进行性能测试，并对比目前其他主流的开源消息中间件产品Kafka、ActiveMQ，记录测试结果，并分析测试数据以检测本发明的消息实时性、吞吐量是否达到设计要求。

由于硬件条件限制，测试所使用的是虚拟机集群搭建测试环境，采用VMware10工具虚拟4台Linux版本为CentOS 6.5的主机。其中ActiveMQ和Kafka各需要其中3台作为broker。测试系统除了3台主机运行broker以外，还需要多布置1台主机运行协调服务。硬件环境如表1所示。

表1硬件环境

所需软件配置如表2所示。

表2软件配置

软件配置操作系统CentOS 6.5Kafka2.10-0.10.0.0版本Zookeeper3.4.8版本ActiveMQ5.8.0版本本消息系统1.0版本JREJava Runtime Environment 6.0

与本专利所对比的三个消息系统都是运行在Java虚拟机上的，所以有必要统一JVM的主要参数，如下所示。

JVM的主要参数：

Java HotSpot(TM)64-Bit ServerVM 1.7.0_67

-XX:UseParallelGC

-Xms:512M

-Xmx:1G

-XX:NewSize:256M

-XX:MaxNewSize:512M

-XX:PermSize:128M

-XX:MaxPermSize:128M

消息实时性测试：k个线程模拟k个消息发布者并发向消息服务器端发送基于不同Topic的大小为1K的消息，同时由k个消息订阅者监听各自订阅的Topic消息，每个线程发送50条消息，记录每条消息从发布者发出到消息被订阅者消费的平均延时。

如图7所示，记录了三个消息系统在16、32、64、128、256个线程并发条件下的消息时延，可以看出本发明在高并发连接情况下的消息时延性能明显好于Kafka和ActiveMQ，原因在于设计良好的通信层、线程模型以及消息推拉模型，能最大限度的优化并发连接。

系统吞吐量测试：分别启动16、32、64、128、256个线程并发发送消息并监听消息的接收，一个线程对应一个Topic，每个线程循环发送50条消息，让测试程序运行一段时间，记录发送并接受成功的消息数量以及总运行时间，然后计算系统TPS(每秒完成消息发送并接受成功的数量)。

如图8所示，在并发量比较小的时候本发明会略低于Kafka的TPS，但随着并发量的提高，本发明TPS会明显上升并超过Kafka。可见，本发明的消息传输机制在高并发访问的情况下，每秒成功处理的消息数量要明显高于Kafka和ActiveMQ。

以上所举实施例，对本发明的目的、技术方案和优点进行了进一步的详细说明，所应理解的是，以上所举实施例仅为本发明的优选实施方式而已，并不用以限制本发明，凡在本发明的精神和原则之内对本发明所作的任何修改、等同替换、改进等，均应包含在本发明的保护范围之内。