Apache Kafka

——学习笔记

Posted by Samuel on July 26, 2017

目录

简介

Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发,之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的,分区的和可复制的提交日志服务。

Apache Kafka与传统消息系统相比,有以下不同:

  • 它被设计为一个分布式系统,易于向外扩展;
  • 它同时为发布和订阅提供高吞吐量;
  • 它支持多订阅者,当失败时能自动平衡消费者;
  • 它将消息持久化到磁盘,因此可用于批量消费,例如ETL,以及实时应用程序。

相关概念

Broker

Kafka集群包含一个或多个服务器,这种服务器被称为broker。一台Kafka服务器就是一个broker,一个集群由多个broker组成,一个broker可以容纳多个topic。

Topic

每条发布到Kafka集群的消息都有一个类别,这个类别被称为Topic,Kafka中Topic可以理解为一个存储消息的队列。物理上不同Topic的消息分开存储,逻辑上一个Topic的消息虽然保存在一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处。

Partition

Partition是物理上的概念,Kafka物理上把Topic分成一个或多个Partition,每个Partition在物理上对应一个文件夹,该文件夹下存储这个Partition的所有消息和索引文件。如创建topic1和topic2两个topic,且分别有13个和19个Partition分区,则整个集群上相应会生成32个文件夹。为了实现扩展性,一个非常大的topic可以分布到多个broker上,但Kafka只保证按一个partition中的顺序将消息发给consumer,不保证一个topic的整体(多个partition间)的顺序。

Producer

负责发布消息到Kafka broker。

Consumer

消息消费者,向Kafka broker读取消息的客户端。

Consumer Group (CG)

这是Kafka用来实现一个topic消息的广播(发给所有的consumer)和单播(发给任意一个consumer)的手段。一个 topic可以属于多个CG。topic的消息会复制(不是真的复制,是概念上的)到所有的CG,但每个CG只会把消息发给该CG中的一个consumer。如果需要实现广播,只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic。每个consumer属于一个特定的Consumer Group, Kafka允许为每个consumer指定group name,若不指定group name则属于默认的group。

特性

  1. 数据在磁盘的存储代价为O(1)
  2. 高吞吐率:普通节点上每秒也能处理成百上千的message
  3. 显式分布式:所有的producer、broker和consumer都会有多个,均匀分布并支持通过Kafka服务器和消费机集群来分区消息。
  4. 支持数据并行加载到Hadoop中
  5. 支持Broker间的消息分区及分布式消费,同时保证每个partition内的消息顺序传输。
  6. 同时支持离线数据处理和实时数据处理:当前很多的消息队列服务提供可靠交付保证,并默认是即时消费(不适合离线),而Kafka通过构建分布式的集群,允许消息在系统中累积,使得Kafka同时支持离线和在线日志处理。
  7. 支持在线水平扩展

架构

最简单的Kafka部署图

如果将消息的发布(publish)称作producer,将消息的订阅(subscribe)表述为consumer,将中间的存储阵列称作broker,这样可以得到一个最简单的消息发布与订阅模型:

image

Kafka的拓扑图

Kafka是显示的分布式消息发布和订阅系统,除了有多个producer, broker,consumer外,还有一个zookeeper集群用于管理producer,broker和consumer之间的协同调用。

image

从图中可以看出,一个典型的Kafka集群中包含若干Producer(可以是web前端产生的PageView,或者是服务器日志,系统CPU、Memory等),若干broker(Kafka支持水平扩展,一般broker数量越多,集群吞吐率越高),若干Consumer Group,以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置,选举leader,以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker,Consumer使用pull模式从broker订阅并消费消息。

图上有个细节需要注意,Producer到Broker的过程是push,也就是有数据就推送到Broker,而Consumer到Broker的过程是pull,是通过Consumer主动去拉数据的,而不是Broker把数据主动发送到Consumer端的。

Zookeeper和Producer、Broker、Consumer的协同工作

为了便于理解,假定此时Kafka集群中有两台Producer,但只有一台Kafka的Broker、Zookeeper和Consumer。如下图所示的部署集群。

image

  1. Kafka Broker其实就是Kafka的server,Broker主要做存储用,每个Broker启动后会在Zookeeper上注册一个临时的broker registry,包含Broker的IP地址和端口号,所存储的topics和partitions信息。
  2. Zookeeper,可以把Zookeeper想象成它维持了一张表,记录了各个节点的IP、端口等配置信息。
  3. Producer1, Producer2, Consumer的共同之处就是都配置了zkClient,更明确的说,就是运行前必须配置Zookeeper的地址,道理也很简单,因为他们之间的连接都是需要Zookeeper来进行分发的。
  4. Kafka Broker和Zookeeper可以放在一台机器上,也可以分开放,此外Zookeeper也可以配集群,这样就不会出现单点故障。
  5. 每个Consumer启动后会在Zookeeper上注册一个临时的consumer registry:包含Consumer所属的Consumer Group以及订阅的topics。每个Consumer Group关联一个临时的owner registry和一个持久的offset registry。对于被订阅的每个partition包含一个owner registry,内容为订阅这个partition的consumer id,同时包含一个offset registry,内容为上一次订阅的offset。

整个系统运行的顺序可简单归纳为:

  1. 启动Zookeeper的server。
  2. 启动Kafka的server。
  3. Producer如果生产了数据, 会先通过Zookeeper找到Broker,然后将数据存放进Broker。
  4. Consumer如果要消费数据,会先通过Zookeeper找对应的Broker,然后消费。

Producer代码实例:

producer = new Producer(...);
message  = new Message("Hello Ebay".getBytes());
set      = new MessageSet(message);
producer.send("topic1", set);

发布消息时,Producer先构造一条消息,将消息加入到消息集set中(Kafka支持批量发布,可以往消息集合中添加多条消息,然后一次性发布), send消息时,client需指定消息所属的topic。

Consumer代码实例:

streams[] = Consumer.createMessageStreams("topic1", 1);
for (message : streams) {
    bytes = message.payload();
    // do something with the bytes
}

订阅消息时,Consumer需指定topic以及partition num(每个partition对应一个逻辑日志流,如topic代表某个产品线,partition代表产品线的日志按天切分的结果),client订阅后,就可迭代读取消息,如果没有消息,client会阻塞直到有新的消息发布。Consumer可以累积确认接收到的消息,当其确认了某个offset的消息,意味着之前的消息也都已成功接收到,此时Broker会更新Zookeeper上的offset registry。

那么怎样记录每个Consumer处理的信息的状态呢?其实在Kafka中仅保存了每个Consumer已经处理数据的offset。这样有两个好处:一是保存的数据量少,二是当Consumer出错时,重新启动Consumer处理数据时,只需从最近的offset开始处理数据即可。

C语言接口librdkafka