一种分布式消息队列的可靠性研究

杨冬晖

摘要：现如今，互联网已经渗透到日常生活的方方面面。生活的各个领域都迎来大数据的影响，数据传输过程的可靠性，以及如何有效地使用数据尤为重要。分布式消息系统可以有效地解决大规模分布式系统中消息传递问题。因此需要一个高吞吐量，高性能以及具有一定可靠性的分布式消息系统。Kafka是一个处理海量数据的分布式消息系统。[1]Kafka具有高效的数据传输速率，相对于其他的消息队列系统具有较高的性能，采用发布/订阅模式。该文主要总结介绍了kafka系统的架构以及特征，重点介绍了分布式集群下kafka如何通过副本模式保证其消息的可靠传输。并通过实验验证其副本模式的可靠性，以支持后续相关研究。

关键词：分布式消息系统；分布式系统；卡夫卡；可靠性

中图分类号：TP311 文献标识码：A 文章编号：1009-3044（2015）21-0075-02

An Research on the Reliability of a Distributed Message Queue

YANG Dong-hui

（School of Software Engineering， Tongji University， Shanghai 201804， China）

Abstract：Nowadays， the Internet has penetrated into every aspect of our daily lives. Every field of social life have the impact of big data transmission， and how to effectively use the data is particularly important. Distributed information system can effectively solve the problem of transmission information in large scale distributed systems. So we need a high throughput， high performance and distributed information system with a certain reliability. Kafka is a distributed information system with a massive data processing. Kafka has efficient data transfer rate， and compared with other message queuing the system has higher performance. Using the publish / subscribe model. This paper mainly introduces the Kafka system structure and characteristics， introduced the Kafka how to ensure reliable transmission of messages with replica. And ensure reliability by experiments， in order to support the follow-up study.

Key words：distributed messaging system； distributed system； Kafka； reliability

在现如今生活中，网上购物已成为日常生活的一部分，电商的发展规模越来越庞大。大型互联网电商公司的业务越来越复杂，不同业务需要不同的子系统进行支持，每个子系统之间的通信需要依靠消息系统支持，这就进一步加大了对消息系统的依赖。当消息系统出现故障，消息不能按时传递、消息丢失等等都会造成不可估量的后果，公司可能造成重大损失。因此分布式系统中的消息通信系统的可靠性尤为重要。

分布式系统[2，11]是指分散的物理机通过互联网连接建立起的一套软件系统，分布式系统具有高度的内聚性和透明性。分布式系统中数据的传输需要面临比传统系统更多的挑战。如何保证分布式系统中大量，高速以及可靠的数据传输是分布式消息队列需要解决的问题。消息队列是操作系统的进程之间用于通信的一种机制，两个或多个进程间通过访问共同的消息队列完成消息的交换。在分布式环境下分布式的消息队列能在客户端和服务端提供同步和异步的连接，实现应用程序之间的协同。

1 KAFKA可靠性研究

1.1 相关背景

在分布式消息队列[3]发展初期，消息传递通常采用点对点的传输结构，发送方需要事先指明接收方的地址，虽然消息的接收方和发送方是松耦合连接的，不必保持相互之间通信的同步。但是由于消息在传递过程中需要绑定接收方的地址，系统不够灵活难以扩展。由于以上的原因分布式消息系统向发布/订阅[4]模式转变，发布/订阅模式具有异步松耦合和多对多通信的特点，此模式适应目前大多数企业分布式计算环境的要求。发布/订阅模式下发送消息的一方称为发布者，接收消息的一方称为订阅者。消息不再被发送到指定的接收者中而是转而发送给一个中间的消息代理服务器。订阅者只需去服务器中接收自己感兴趣的消息，发布者不需要知道何种订阅者接收了发布的消息。这种发布/订阅的模式更易于扩展。常见的发布/订阅消息队列包括Kafka[5]、RabbitMQ[6]、ActiveMQ[7]以及Microsoft MSMQ等。其中RabbitMQ是使用Erlang语言编写的一个开源的消息队列，是AMQP协议的一个实现。它实现了代理（Broker）架构，使得消息在发送到客户端之前可以在中央节点上排队。

1.2 Kafka结构

Kafka由Linkedin公司开发，并使用Scala语言编写。Kafka是一个分布式的、分区的、多副本的、多订阅者的日志提交系统[9]。Kafka系统包括消息的生产者（producer），消息的消费者（consumer），消息代理者（broker）和管理者（zookeeper） [8]四个部分。生产者生产的消息被放在topic中。一个topic中可以设置多个partition，每一个partition可以对应一个消费者被消费。当设置了多个partition，生产者在生产消息的时，需指定对应的partition，如果没有指定，则会使用默认值。每个partition是一个有序的、不可变的消息序列，这个序列可以被连续地追加。每条消息会有一个序列号，在文件中的位置被称为offset（偏移量），offset为一个long型数字，唯一标识一条消息。生产者生产消息，产生的消息发送到broker，等待消费者的接收。消息在broker中根据topic和partition来区分不同。消费者根据topic和partition的值从broker中订阅消息。消费者在想要订阅消息时向broker端发送请求，告知其topic和partition值来获取消息。消费者可以决定从partition何处开始消费，并通过重置offset的值来重新消费已经消费的消息。在Kafka中消费者在和broker建立连接后，主动拉取消息。生产者和消费者之间消息的传递可以用图1所示[9]。

管理者zookeeper在其中进行协调控制，管理broker和consumer的动态加入和离开，维护了生产者和消费者之间的关系和topic中的信息，并提供一定的负载均衡支持。Kafka server为生产者消费者提供服务，当生产者和消费者运行时，都需要配置相应的zookeeper信息。Zookeeper服务器主要功能为与kafka服务器进行交互，zookeeper服务器对kafka服务器进行简单的管理。每个broker启动后会在zookeeper上临时注册，注册信息包含broker的ip地址和端口号，broker上存在的topic和此topic的partition。在kafka中producer可以将生产的消息指定发送到某一个具体的partition。相对应的对于consumer来说，每一个consumer属于一个消费者集群，每个group中的consumer消费消息相互独立。

1.3 Kafka可靠性研究

Kafka消息队列为分布式系统提供一定的可靠性保证，针对生产者生产的消息，在集群模式下，kafka可以将每一个partition中的数据复制到多个kafka server中。当复制为多个副本时，每个partition会由zookeeper指定一个leader节点和多个follower节点（当副本数多于2个）。备份的个数可以通过修改broker配置文件或者手动配置更改。Follower节点需要和leader节点保持同步。其中leader节点负责处理读写请求，leader节点还需要监控它所控制的所有follower节点的运行状态，当follower节点与leader节点严重不一致或者节点失效，leader节点会删除此follower节点。当leader节点失效时，通过选举将在存活的follower节点中重新选举出新的leader节点，新的leader节点替代失效leader节点继续负责监控整个集群运行。因此当分布式集群中某一个业务节点失效时，只要还有一个节点存活，即还有一个partition的副本。此消息都可以进行正常发送和接收[10]。

Kafka集群的可靠性通过partition的多副本方式得到提高。在kafka集群的运行过程中，每一个kafka server上可以运行多个生产者或多个消费者，kafka server启动时会在zookeeper上注册相应的信息。对相应的partition保存了多个副本时，当其中某个kafka server发生故障，zookeeper会将失效的kafka server部署到集群中处于存活状态的kafka server中去完成此操作后，分布式集群中消息通信会通过备用的kafka server进行传递，由此保证消息通信服务不受影响。

Kafka集群的建立使得partition可以复制为多个副本，当正在使用的kafka server因故障而停止工作时，可以使用副本继续进行生产者消费者的消息之间的传递，不需要重启kafka server，保证了进程的正常运行。如图2所示，当kafka服务开始时，生产者与消费者进程正常运行，经过broker进行消息的传递，生产者将消息推送给已订阅的消费者。Zookeeper server对kafka server进行管理，本图包含两个kafka server所以topic最多保存了两个副本。如图中所示，正在提供服务的kafka server出现故障，zookeeper server选举新的leader节点并调用其副本继续提供服务。生产者和消费者正常的消息传递会通过副本继续进行，消息不会丢失，保证了分布式集群中消息的可靠传递。

2 实验分析

通过实验验证分析kafka分布式集群的可靠性。实验环境为三台kafka服务器与一台zookeeper服务器。实验过程为：

1）启动zookeeper服务与3台kafka服务；

2）创建1个生产者与1个消费者，并创建1个topic；

3）针对创建的topic将其设置为3个副本模式，并观察哪一个节点为leader节点，哪两个节点为follower节点；

4）生产者发送信息，手动终止leader节点kafka服务，观察消费者有无正常接收消息；

5）手动终止还存活的kafka服务节点中的一个，观察消费者是否可以正常接收消息；

6）手动终止最后一个存活的kafka服务节点，观察消费者是否可以正常接收消息。

通过上述实验，得到实验结果如下：当第一次终止leader节点后，zookeeper会重新选举出新的leader节点，生产者发布消息后，订阅此topic的消费者仍可以正常的接收消息。当终止的 服务节点不为leader节点时，leader节点不会重新选举，消费者仍可接收订阅的消息。当终止所有kafka服务节点时，生产者生产的消息无法发布到broker中，消费者也无法从broker中订阅消息。

通过上述实验结果，可以得出kafka分布式集群通过多副本，选取leader节点对多个副本进行管理的形式，保证集群消息传递具有一定的可靠性。

3 结束语

分布式消息队列在如今互联网企业得到越来越多的应用。Kafka为分布式集群中消息传递提供了一定的可靠性保证。本文分析了kafka的基本结构，并对其副本模式进行了分析。实验验证了其副本模式下生产者消费者消息的传递是否能容忍一定的故障。为后续深入研究kafka提供前期支持。

参考文献：

[1] Goodhope K K G， Goodhope K. Building LinkedIn's Real-time Activity Data Pipeline[J]. Bulletin of the Technical Committee on Data Engineering， 2012（2）.

[2] 胡华平， 金士尧， 王召福. 分布式系统的可信性研究[J]. 计算机工程与科学， 1998（1）： 48-53.

[3] Patel D， Khasib F， Sadooghi I， et al. Towards In-Order and Exactly-Once Delivery Using Hierarchical Distributed Message Queues[C]//IEEE/ACM International Symposium on Cluster， Cloud and Grid Computing. IEEE， 2014：883-892.

[4] 马建刚， 黄涛， 汪锦岭， 等. 面向大规模分布式计算发布订阅系统核心技术[J]. 软件学报， 2006， 17（1）： 134-147.

[5] Jay Kreps， Neha Narkhede， Jun Rao. Kafka： a Distributed Messaging System for Log Processing[Z].2011.

[6] 袁佳， 郭燕慧. 基于rabbitmq的海量日志的分布式处理[J]. 软件， 2013（7）： 19-23.

[7] 戴俊， 朱晓民. 基于 ActiveMQ 的异步消息总线的设计与实现[J]. 计算机系统应用， 2010（8）： 254-257.

[8] Okorafor E， Patrick M K. Availability of Jobtracker machine in hadoop/mapreduce zookeeper coordinated clusters[J]. Advanced Computing： An International Journal （ACIJ）， 2012， 3（3）： 19-30.

[9] Apacher Kafka[EB/OL]. http：//kafka.apache.org/.

[10] Chen J， Arumaithurai M， Fu X， et al. Reliable publish/subscribe in content-centric networks[C]//Proceedings of the 3rd ACM SIGCOMM workshop on Information-centric networking. ACM， 2013： 21-26.

[11] 马浩然. 基于NS3的分布式消息系统Kafka的仿真实现[J]. 软件， 2015（1）.