第一部分:初步了解Kafka

Apache Kafka是一个分布式流媒体平台。这到底是什么意思？

流媒体平台具有三个关键功能：

发布和订阅记录流，类似于消息队列或企业消息系统。

以容错的持久方式存储记录流。

记录发生时的处理流。

卡夫卡通常用于两类广泛的应用：

1.建立实时流数据管道，在系统或应用之间可靠地获取数据

2.建立对数据流进行转换或反应的实时流应用程序

为了理解卡夫卡是如何做到这些的，让我们从下到上探究卡夫卡的能力。

首先是几个概念：

卡夫卡作为一个集群运行在一个或多个可以跨多个数据中心的服务器上。

卡夫卡集群将记录流存储在称为主题的类别中。

每个记录由一个键、一个值和一个时间戳组成。

Kafka 有四个核心API：

生产者API(Producer API)允许应用程序将记录流发布到一个或多个卡夫卡主题。

消费者API(Consumer API)允许应用程序订阅一个或多个主题，并处理产生给它们的记录流。

流API(Streams API)允许应用充当流处理器，消耗来自一个或多个主题的输入流，并将输出流生成到一个或多个输出主题，有效地将输入流转换为输出流。

连接器API(Connector API)允许建立和运行可重用的生产者或消费者，将卡夫卡主题连接到现有的应用程序或数据系统。例如，关系数据库的连接器可能会捕获表的每一个更改。

在卡夫卡中，客户端和服务器之间的通信是用简单、高性能、语言不可知的TCP协议来完成的。该协议版本化，并保持与旧版本的向后兼容性。我们为卡夫卡提供Java客户端，但客户端可以用多种语言提供。

Topics and Logs

让我们先深入研究核心抽象卡夫卡为一条记录流提供的主题。

主题是记录发布的类别或进给名称。卡夫卡中的主题总是多用户的，也就是说，主题可以有一个或多个订阅该数据的用户。

对于每个主题，卡夫卡集群维护一个看起来像这样的分区日志：

每一个分区都是一个有序的、不可变的记录序列，它连续地附加到结构化提交日志。每个分区中的记录都被分配一个连续的ID号，称为唯一地标识分区内的每个记录的偏移量。

无论是否已经使用可配置的保留周期消耗，所有的已发布记录都能持久地保存卡夫卡集群。例如，如果保留策略设置为两天，那么在发布记录后的两天，它可以用于消费，之后将被丢弃以释放空间。卡夫卡的性能相对于数据大小是有效不变的，所以长时间存储数据不是问题。

事实上，在每个消费者基础上保留的唯一元数据是该用户在日志中的偏移或位置。这种偏移是由消费者控制的：正常情况下，消费者会在读取记录时线性地偏移其偏移量，但事实上，由于该位置是由消费者控制的，所以它可以按它喜欢的任何顺序消耗记录。例如，消费者可以重置为旧的偏移量，以重新处理过去的数据，或者跳过最近的记录并开始从“现在”开始消费。

这种组合的特征意味着卡夫卡的消费者非常便宜，他们可以来来往往，对集群或其他消费者没有太大的影响。例如，您可以使用我们的命令行工具来“尾随”任何主题的内容，而不改变任何现有消费者所消耗的内容。

日志中的分区有多种用途。首先，它们允许日志超出一个适合于单个服务器的大小。每个单独的分区必须适合于承载它的服务器，但是一个主题可能有许多分区，因此它可以处理任意数量的数据。其次，它们作为并行的单位更多的是在这一点上。

Distribution

日志分区被分布在卡夫卡集群中的服务器上，每个服务器处理数据并请求共享分区。每个分区在可配置的多个服务器上复制以容错。

每个分区都有一个服务器充当“领导者”和零个或多个服务器，充当“追随者”。领导者处理所有的读写请求的分区，而追随者被动复制的领导者。如果领导者失败，其中的一个追随者将自动成为新的领导者。每个服务器充当一些分区的领导者和其他人的追随者，因此集群内的负载很好地平衡。

Geo-Replication

Kafka MirrorMaker为您的集群提供地理复制支持。使用镜像机，消息在多个数据中心或云区域上被复制。可以在主动/被动场景中使用此备份和恢复；或者在活动/活动场景中，将数据更靠近用户，或支持数据位置要求。

Producers

生产者将数据发布到他们选择的主题中。生产者负责选择要分配给主题内的哪个分区的记录。这可以在循环的方式下完成，只是为了平衡负载，或者它可以根据一些语义划分函数来完成（例如基于记录中的某些键）。

Consumers

消费者用消费者组名称来标记自己，并且发布到主题的每个记录被传递到每个订阅消费者组中的一个消费者实例。消费者实例可以在单独的进程中或在单独的机器上。

如果所有的消费者实例都具有相同的消费群，那么记录将有效地在消费者实例上进行负载均衡。

如果所有的消费者实例都有不同的消费群体，那么每个记录将被广播到所有的消费过程。

两个服务器卡夫卡集群托管四个分区（P0至P3）与两个消费群体。消费者组A有两个消费者实例，B组有四个。

然而，更常见的是，我们发现话题有少量的消费群体，每个都有一个“逻辑用户”。每个组由许多可扩展性和容错性的消费者实例组成。这仅仅是发布订阅语义，其中订阅服务器是一组消费者而不是单个进程。

在卡夫卡中实现消费的方法是通过将日志中的分区除以消费者实例，使得每个实例在任何时间点都是分区的“公平共享”的唯一消费者。这个组中的成员保持过程是由卡夫卡协议动态处理的。如果新实例加入组，它们将从组的其他成员接管一些分区；如果一个实例死亡，它的分区将被分发到其余实例。

卡夫卡只提供一个分区内的记录的总顺序，而不是在一个主题中的不同分区之间。每一个分区排序结合能力分区数据的关键是足够的大多数应用程序。但是，如果需要对记录进行总排序，则可以用只有一个分区的主题来实现，但这将意味着每个消费者组只有一个消费者进程。

多租户技术

可以将卡夫卡部署为多租户解决方案。多租户是通过配置哪些主题可以产生或消耗数据来实现的。也有对配额的操作支持。管理员可以在请求上定义和执行配额，以控制客户端使用的代理资源。

保证

在高级别卡夫卡提供以下保证：

生产者发送给特定主题分区的消息将按照发送的顺序添加。也就是说，如果记录M1是由与记录M2相同的生产者发送的，M1是最先发送的，那么M1将具有比M2更低的偏移量，并且在日志中出现得更早。

一个用户实例以记录存储在日志中的顺序查看记录。

对于具有复制因子N的主题，我们将容忍多达N-1服务器故障而不丢失提交到日志的任何记录。

Kafka可以保证同一个分区里的消息是有序的,也就是说生产者按照一定的顺序发送消息,broker就会按照这个顺序把他们写入分区,消费者也会按照同样的顺序读取他们

卡夫卡作为消息传递系统

卡夫卡的流概念如何与传统的企业消息系统相比较？

消息传递传统上有两种模式：排队和发布订阅。在队列中，消费者池可以从服务器读取，并且每个记录都转到其中一个；在发布订阅中，记录被广播给所有消费者。这两种模式各有其优点和缺点。排队的优点在于，它允许您在多个消费者实例上划分数据的处理，这使得您可以对处理进行缩放。不幸的是，一旦一个进程读取了它的数据，队列就不再是多用户。发布订阅允许您将数据广播到多个进程，但是由于每个消息都流向每个订阅服务器，所以无法进行缩放处理。

卡夫卡的消费群体概念概括了这两个概念。与队列一样，消费者组允许您在进程集合（消费者组的成员）上划分处理。与发布订阅一样，卡夫卡允许您向多个用户组广播消息。

卡夫卡模型的优点是，每个主题都具有这些属性，它可以缩放处理，并且也是多用户，不需要选择一个或另一个。

卡夫卡比传统的消息传递系统具有更强的排序保证。

传统队列在服务器上保留记录顺序，如果多个用户从队列中消耗，则服务器按其存储的顺序分发记录。然而，尽管服务器按顺序分发记录，但是记录是异步传送给消费者的，因此它们可能会在不同的消费者之间无序地到达。这实际上意味着在并行消耗的存在下记录的顺序丢失。消息传递系统经常围绕着这一点而工作，它有一个“独占消费者”的概念，它只允许一个进程从队列中消耗，但这当然意味着在处理过程中没有并行性。

卡夫卡做得更好。通过在主题中具有并行性分区的概念，卡夫卡能够在消费者进程池中提供排序保证和负载平衡。这是通过将主题中的分区分配给消费者组中的消费者来实现的，这样每个分区就被该组中的一个消费者完全消耗掉。通过这样做，我们确保消费者是该分区的唯一读取器，并按顺序消耗数据。由于有许多分区，这仍然平衡了许多消费者实例的负载。但是注意，消费者组中的消费者实例不能多于分区。

卡夫卡作为存储系统

任何允许发布消息的消息队列与它们之间的解耦都有效地充当了飞行消息的存储系统。卡夫卡的不同之处在于它是一个非常好的存储系统。

写入卡夫卡的数据被写入磁盘并复制用于容错。卡夫卡允许生产商等待确认，这样写才被认为是完整的，直到它被完全复制，并保证即使服务器写入失败也会坚持。

磁盘结构卡夫卡使用规模井卡夫卡将执行相同的，无论你有50 KB或50 TB的持久性数据在服务器上。

作为认真对待存储并允许客户端控制其读取位置的结果，您可以将卡夫卡视为专用于高性能、低延迟提交日志存储、复制和传播的专用分布式文件系统。

用于流处理的卡夫卡

仅仅读取、写入和存储数据流是不够的，其目的是实现对流的实时处理。

在卡夫卡中，流处理器是从输入主题获取连续数据流的任何东西，对该输入执行一些处理，并产生连续的数据流到输出主题。

例如，零售应用程序可能会接收销售和出货的输入流，并输出从该数据计算的重新排序和价格调整的流。

可以直接使用生产者和消费者API来进行简单的处理。然而，对于更复杂的转换，卡夫卡提供了一个完全集成的流API。这允许构建非平凡处理的应用程序，它们可以计算流中的聚合或一起加入流。

这个工具有助于解决这类应用程序面临的难题：处理无序数据、重新处理输入、代码更改、执行状态计算等。

流API构建在卡夫卡提供的核心原语上：它使用生产者和消费者API来进行输入，使用卡夫卡进行状态存储，并在流处理器实例中使用相同的组机制来容错。

把碎片拼在一起

消息传递、存储和流处理的这种组合看起来是不寻常的，但这对于卡夫卡作为流媒体平台的角色是至关重要的。

分布式文件系统（如HDFS）允许存储静态文件以进行批量处理。实际上，这样的系统允许存储和处理过去的历史数据。

传统的企业消息系统允许处理订阅后到达的未来消息。以这种方式构建的应用程序在到达时处理未来的数据。

卡夫卡结合了这两种能力，组合对于卡夫卡作为流媒体应用平台以及流数据管道来说都是至关重要的。

通过结合存储和低延迟订阅，流式应用程序可以以相同的方式处理过去和将来的数据。也就是说，单个应用程序可以处理历史、存储的数据，而不是在到达最后记录时结束，它可以在将来的数据到达时保持处理。这是包含批处理和消息驱动应用程序的流处理的一般概念。

同样，对于流式数据管道，订阅到实时事件使得使用卡夫卡用于非常低延迟的流水线是可能的；但是，可靠地存储数据的能力使得它可以用于关键数据，其中必须保证数据的传递或整合。n具有离线系统，只周期性地加载数据，或者可以在长时间内进行维护。流处理设施使得在到达时转换数据成为可能。

第二部分:Kafka技术进阶

生产者概括

Kafka发送消息的主要步骤

我们从创建一个ProducerRecord对象开始,ProducerRecord对象需要包含目标主题和要发送的内容.我们还可以指定键分区,在发送ProducerRecord 对象时,生产者要先把键和值对象序列化成字节数组,这样他们才能在网络上传输. 接下来,数据被传给分区器,如果之前在ProducerRecord对象里指定了分区,那么分区器就不会再做任何事情,直接把指定的分区返回.如果没有指定分区,那么分区器就会根据ProducerRecord对象的键来选择一个分区.选好分区以后,生产者就知道该往哪个主题和分区发送这条消息了. 紧接着,这条记录被添加到一个记录批次里,这个批次里的所有消息会被发送到相同的主题和分区上.有一个独立的线程负责把这些记录批次发送到相应的broker上. 服务器在收到这些消息时会返回一个响应.如果消息成功写入Kafka,就返回一个RecordMetaData对象,它包含了主题和分区信息,以及记录在分区里的偏移量.如果写入失败,则返回一个错误.生产者在收到错误之后会尝试重新发送消息,几次之后如果还是失败,就返回错误信息.

创建Kafka生产者

要往Kafka写入消息,首先要创建一个生产者对象,并设置一些属性,Kafka生产者有3个必选属性(什么意思不作说明)

bootstrap.servers

key.serializer

value.serializer

其他默认属性:

acks

buffer.memory

compression.type

retries

batch.size

linger.ms

client.id

max.in.flight.requests.oer.connection

timeout.ms,request.timeout.ms和metadata.fetch.timeout.ms

max.block.ms

max.request.size

resive.buffer.bytes和send.buffer.bytes

下面代码片段演示了如何创建一个新生产者,这里只指定了必要的属性,其他使用默认设置

private Properties kafkaProps = new Properties(); kafkaProps.put("bootstrap.servers","broker1:9092,broker2:9092");kafkaProps.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");kafkaProps.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");producer = new KafkaProducer
     
      (kafkaProps);
     

上述代码片段主要意思是 先新建一个Properties对象

因为我们打算把键和值定义成字符串类型,所以使用内置的StringSerializer 在这里我们创建一个新的生产者对象,并为键和值设置了恰当的类型,然后把Properties对象传给他

实例化生产者对象后,接下来就可以开始发送消息了.发送消息主要有以3种方式,

发送并忘记(fire-and-forger):我们把消息发送给服务器,但并不关心他是否正常到达.大多数情况下消息会正常到达,因为Kafaka是高可用的,而且生产者会自动尝试重发,不过使用这种方式有时候也会丢失一些消息

同步发送send():我们使用send()方式发送消息,他会返回一个Future对象,调用get()方法进行等待,就可以知道消息是否发送成功

异步发送send():我们使用send()方式发送消息,并指定一个回调函数,服务器在返回响应时调用该函数

上面的例子使用的都是单线程,但其实生产者是可以使用多线程来发送消息的刚开始的时候可以使用单个消费者和单个线程.如果需要更高的吞吐量,可以在生产者数量不变的情况下增加线程数量.如果这样还不够,可以增加生产者数量

发送消息到Kafka

最简单消息发送方式如下所示

ProducerRecord
     
       record = new ProducerRecord<>("CustomerCountry":"Precision Products","France");try{     proucer.send(record);} catch (Exception e){         e.printStackTrance();}
     

生产者的send() 方法将ProducerRecord 对象作为参数,所以我们要先创建一个ProducerRecord对象.ProducerRecord有多个构造函数,这里只使用其中的一个,他需要目标主题的名字 和要发送的键和值对象 ,他们都是字符串.键和值对象的了新鲜感必须与序列化器的生产者对象相匹配

我们使用生产者的send()方法发送ProducerRecord对象.从生产者的架构图可以看到,消息是先被放进缓冲区,然后使用单独的线程发送到服务器端.send()方法会返回一个包含RecordMetadata的Future对象,不过我们会忽略返回值,所以无法知道消息是否发送成功,如果不关心发送结果,那么可以使用这种发送方式.比如 记录Twitter消息日志,或记录不太重要的应用程序日志

同步发送

最简单的同步发送消息方式如下所示

ProducerRecord
     
       record = new ProducerRecord<>("CustomerCountry","Precision Products","France");try{  producer.send(record).get();} catch (Exception e){       e.printStackTrance();}
     

在这里 producer.send() 方法先返回一个Future对象,然后调用Future对象的get()方法等待Kafka响应,如果服务器返回错误,get()方法会抛异常.如果没有发生错误,我们会得到一个RecordMetadata对象 可以用他来获取消息的偏移量

如果在发送数据之前或在发送过程中发生了任何错误.比如broker返回一个不允许重发消息的异常或者已经超过了重发的次数,那么就会抛出异常.我们只是简单的把异常信息打印出来

异步发送

假设消息在应用程序和Kafka集群之间一个来回需要10ms. 如果在发送完每个消息后都等待回应,那么发送100个消息需要1秒 但如果只发送消息而不等待响应,那么发送100个消息所需要的时间会少很多.大多数时候,我们并不需要等待响应,尽管Kafka会把目标主题,分区信息和消息的偏移量发送回来,但对于发送端的应用程序来说不是必需的.不过在遇到消息发送失败时,我们需要抛出异常,记录错误日志,或把消息写入错误消息文件

为了在异步发送消息的同时能够对异常情况进行处理,生产者提供了回调支持,下面是使用回调的一个例子

private class DemoProducerCallback implements Callback {@Overridepublic void onCompletion(RecordMetadata recordMetadata,Exception e){if (e != null){ e.prinStackTrace();    }  }}ProducerRecord
     
       record =  new ProducerRecord<>("CustomerCountry","Biomedical Materials","USA");producer.send(record,new DemoProducerCallback());
     

为了使用回调,需要实现org.apach.kafka.clients.producer.Callback接口的类,这个接口只有一个onCompletion方法

如果 Kafka返回一个错误 onCompletion方法会抛一个非空(non null)异常 在发送消息时传进去一个回调对象

消费者(KafkaConsumer)

在了解如何从Kafaka读取消息之前,我们先先了解一下消费者和消费者群组的概念

假设我们有一个应用程序要从一个Kafka主题读取消息并验证这些消息,然后把他们存储起来,应用程序需要创建一个消费者对象,订阅主题并开始接收消息,然后验证消息并保存结果,过了一阵子 生产者让主题写入消息的速度超过了应用程序验证数据的速度,这个时候该怎么办?如果只使用单个消费者处理消息,应用程序永远跟不上消息的生成速度,这个时候就需要像多个生产者可以向相同的主题写入消息一样,我们也需要使用多个消费者从同一个主题读取消息,对消息进行分流.Kafka 消费者从属于消费者群组,一个群组里的消费者订阅的是同一个主题,每个消费者接收主题一部分分区的消息,分区的一个主题消息会被不同消费组订阅,一个消息只能被每个消费者群组中的一个消费者接收.

消费者群组和分区再均衡

群组里的消费者共同读取主题的分区,一个新的消费者加入群组时,他读取的是原本由其他消费者读取的消息.当一个消费者被关闭或者发生崩溃时,他就离开群组,原本由他读取的分区将有群组里的其他消费者来读取,在主题发生变化时,比如管理员添加新的分区,会发生分区重分配.分区的所有权从一个消费者转移到另外一个消费者这样的行为称作再均衡

程序如何触发再均衡? 消费者通过向群组协调器的broker发送心跳来维持他们和群组的从属关系以及他们对分区的所有权关系,只要消费者可以以正常的的时间间隔发送心跳就被认为是活跃的说明他还在读取分区的消息.消费者会在轮询消息或者提交偏移量时发送心跳.如果消费者停止发送心跳的时间足够长,会话过期,群组协调器认为他已经死了,就会触发一次再均衡.

创建Kafka消费者

{   Properties props = new Properties();   props.put("bootstrap.servers","broker1:9092,broker2:9092");   props.put("group.id","CountryCounter");   props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");   props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer
     
       consumer = new KafkaConsumer
      
       (props);}
      
     

订阅主题

consumer.subscribe(Collection.singletonList("customerCountries"));//主题名"customerCountries"

消费者配置

fetch.min.bytes

fetch.max.wait.ms

max.partition.fetch.bytes

session.timeout.ms

auto.offset.reset

enable.auto.commit

partition.assignment.strategy

client.id

max.poll.records

receive.buffer.bytes/send.buffer.bytes

到这里Kafka的基础内容已经介绍完了 ,如果想深入了解这些是远远不够的 在这里可以推荐几本书给大家

如果想对Kafka的整体有深刻的认识可以读<<Kafka权威指南>>必读 其次就是 <<Kafka技术内幕>> 这两本书读完 几本就OK了 最后也可以读<<Apache Kafka 源码剖析>>不建议读 当然官网读英文文档最好了