【.com原创文章】ApacheKafka被誉为流行的企业级消息系统。它的初衷是一个发布和订阅记录流的分布式流系统。完美的容错效果受到业内人士的青睐。图片来自Pexels纵览当前数据为王的时代，了解更多ApacheKafka及其常用部署应用。快速实现数据架构（KafkaFastDataArchitecture）已是大势所趋，刻不容缓。下面详细介绍Kafka架构、四大核心API、典型应用场景、Kafkabroker和消息主题、集群创建、StreamAPI（流API）及其处理方式。Kafka：分布式流平台Kafka是一个分布式流平台，用于发布和订阅消息流（也称为记录流或数据流），快速有效地利用I/O对数据流进行批处理、压缩和解耦，以及流数据进入数据湖、应用程序和实时流分析系统。Kafka跨多个服务器复制主题消息分区，允许用户通过自己的应用程序处理这些记录。Kafka的四大核心APIKafka由记录、主题、消费者、生产者、代理、日志、分区和集群组成。一个Kafka主题是一个记录流，每个主题都有一个对应的日志，是该主题在磁盘上的存储，每个主题日志又分为多个分区和分片。KafkaProducerAPI用于生成数据记录流。KafkaConsumerAPI用于消费来自Kafka的记录流。Broker是运行在Kafka集群中的Kafka服务器，由多个代理服务器组成。①生产者API（ProducerAPI）：消息的生产者，向Kafkabroker发送消息的客户端。允许客户端连接到集群中运行的Kafka服务器，并将记录流发布到一个或多个Kafka主题（消息主题）。一个Kafka服务器就是一个broker，一个集群由多个broker组成，一个broker可以容纳多个topic。②消费者API（ConsumerAPI）：消息消费者，从Kafkabroker获取消息的客户端。允许客户端连接运行在集群中的Kafka服务器，消费一个或多个Kafka主题（消息主题）的记录流。③流API（StreamAPI）：充当流处理器，进行输入流和输出流的转换。允许客户端充当流处理器，消费来自一个或多个主题（消息主题）的输入流，并产生输出流输出到一个或多个其他主题（消息主题），从而有效地将输入流转换为输出流。④连接器API（ConnectorAPI）：允许编写可重用的生产者和消费者代码。我们可以从任何关系数据库中读取数据并将其发布到主题，也可以从该主题“消费”数据并将其写入关系数据库。可以看出，ConnectorAPI支持构建和运行可重用的生产者或消费者，并将主题连接到现有的应用程序或数据系统。（比如关系型数据库，它们的连接器可以捕捉到每张表的每一个变化。）Kafka应用场景消息系统Kafka作为一个企业消息系统，通过源系统和目标系统的分离来实现数据交换.与JMS相比，Kafka既有高吞吐的分区，也有高可靠的容错复制。网站活动跟踪跟踪记录用户在网站上的所有事件信息，用于数据分析和离线处理。日志汇总用于处理来自不同系统的日志，尤其是微服务架构的分布式环境中的日志。这类系统通常部署在不同的主机上，因此Kafka需要汇总来自不同系统的各种日志，然后将这些日志集中处理进行分析。指标收集Kafka可用于收集和监控来自各种系统/网络的指标。Kafka配备了专门的指标报告生成工具，如Ganglia、Graphite等。KafkaBrokers&KafkaTopicsKafkaBroker（代理服务器）Kafka集群中的一个实例称为代理（服务器）。在一个Kafka集群中，只要连接到任何一个proxy（服务器），就可以访问整个集群。由ID识别。Kafka主题（MessageTopic）消息主题（Topic）是一条消息记录发布后的逻辑名称。在Kafka中，一个Topic被划分为若干个分区（Partitions）进行消息发布。这些分区分布在集群的各个代理服务器（Brokers）中。为了实现可扩展性，一个非常大的Topic通常分布在多个代理服务器（Brokers）中。由于一个Topic可以分为多个分区（Partition），每个分区（Partition）就是一个有序队列。分区（Partition）中的每条消息都会被分配一个有序的ID（即偏移量，Offset）。如下图所示，假设当前有一个主题（Topic），主题（Topic）有3个分区，集群中有3个代理（Brokers），每个代理有一个分区。要发布到分区的数据以偏移增量附加。其中，“Offset”为偏移量。Kafka的存储文件是按照“offset.kafka”命名的。Offset方法用于方便搜索。如果要找到2046的位置，只需要找到2045.kafka的文件即可。能。以下是使用Partitions时需要注意的几点：每个消息主题（Topic）通过名字来标识，集群中允许有多个命名的消息主题。每条消息的顺序有效性仅限于当前分区级别（在分区级别维护），而不是跨主题。数据一旦写入分区，就不会被覆盖。这就是Kafka中强调的数据不变性（immutability）。分区中的消息以键（keys）、值（values）、时间戳（timestamps）的形式存储在一起。Kafka确保给定键的每条消息都发布到同一个分区。在Kafka集群中，每个分区都有一个领导者，负责对该分区执行读/写操作。上图就是一个例子。当前集群中只有一个消息主题（Topic），包含三个分区（partition0、partition1、partition2），集群中有三个代理服务器（broker1、broker2、broker3）。目前，每个分区的副本被复制到另外两个代理服务器（Broker），即每个代理服务器（Broker）包含三个分区。因此，即使其中两个代理服务器（Broker）发生故障，也不用担心数据丢失。如上，我们在Kafka中创建主题时，总是建议保证主题（Topic）的复制因子大于1且小于/等于集群中代理服务器（Broker）的数量，即非常推荐。在上面的例子中，当前topic的复制因子为3（即一份原始数据，两份副本数据），不难推导出每个分区的bootstrap加上其副本数为“总共3"。在此示例中，每个分区都有一个领导者（称为“领导者”）和另外两个同步副本（称为“跟随者”）。对于partition0，broker1是“leader”，broker2和broker3都是“follower”，所以partition0的所有读写操作都会在broker1中进行。同时更新的内容也会同步复制到broker2和broker3对应的分区中。创建Kafka集群——Demo下面以上图中三个Broker组成的Kafka集群为例，拆解一下创建Kafka集群的步骤。①Kafka集群环境准备首先需要准备一台安装了Zookeeper的机器，没有Zookeeper，Kafka集群将无法运行。同时建议直接从官网下载最新版本的ApacheKafka。当前版本更新到2.11。解压后放在bin目录下：https://archive.apache.org/dist/kafka/1.0.0/kafka_2.11-1.0.0.tgz然后启动ZooKeeper，为什么需要Zookeeper？这里主要负责协调服务，管理broker服务Broker，确定各个partition中的bootstrap，并在Kafka消息主题或broker服务发生变化时及时发出警告。一个Zookeeper实例可以通过以下命令启动：②启动KafkaBrokers安装Kafka成功并启动ZooKeeper实例后，接下来就可以启动KafkaBroker了，这里一共启动了三个KafkaBrokers。具体启动方法：首先定位到Kafka根目录下的“config”文件夹，找到“server.properties”文件，复制三遍。然后分别命名为server_1.properties、server_2.properties、server_3.properties，编辑三个文件内容如下，直接保存：保存后，通过命令打开三个代理服务：③创建主题，创建主题消息使用如下命令Topic：④生成一个bootstrap服务通过Kafka控制台生成器（Kafkaconsole）指定任意代理服务地址，根据之前创建的topic发布一些消息。这个指定的代理服务被认为是一个引导服务程序，用于访问整个集群。⑤“消费”消息是通过Kafka控制台使用的，用户（即：消息消费者）需要指定任意一个代理服务（Broker）地址作为引导服务器。在阅读消息时，用户（即消息消费者）看不到消息的顺序。如上所述，消息的顺序仅在分区级别维护，而不是主题级别。.可以通过以下命令来描述题目，查看各个分区的分布情况，以及各个分区的bootserver：从上面的执行结果可以看出：broker-1是partition0的bootserver.broker-2是分区1的引导服务器broker-3是分区2的引导服务器broker-1,broker-2,broker-3每个分区都有一个副本（同步并相互备份）.KafkaStreamsAPIKafka经常被用来将流式数据实时传输到其他系统。此时Kafka作为中间层，主要用于实时数据管道的解耦和分离。KafkaStream是Kafka生态系统的一部分，提供实时分析能力，支持将流数据传输到大数据平台或RDBMS、Cassandra、Spark，以供日后数据分析。KafkaStreamAPI简单易用，通过其强大的技术能力可以处理存储在其中的所有数据。同时API也为我们提供了一套Kafka标准类的实现规则。为了能够在实际工作中创建支持核心业务的实时应用，我们需要KafkaStreamAPI的大力协助。KafkaStreamAPI的独特之处在于，用它构建的应用程序是普通应用程序。因此，这些应用程序可以像任何其他应用程序一样进行打包、部署和监控，而无需单独安装专用处理集群或类似基础设施，而部署额外的基础设施通常成本很高。流（Stream）是KafkaStreams提供的最重要的抽象对象，它代表了一个无限的、不断更新的数据集。流是不可变数据记录的序列，具有有序性、可重复性和容错性等特点。我们可以简单的把它看成一个记录流（定义为：KStream）或者一个变更日志流（定位为：KTable或者GlobelKTable）。流处理器（StreamProcessor）是处理器拓扑结构中的一个节点，它包含应用于流数据的处理逻辑，一系列的节点组成了拓扑结构中的处理步骤（用于转换数据）。KafkaStreamsAPI处理数据——DemoKafkaStreamAPI在Kafka中提供了流数据处理的两种选择，即消息消费和回写：AdvancedKafkaStreamsDSL（高级DSL）。低级处理器API：用于数据的基本处理、复合处理、本地状态存储。①高层DSL（high-levelDSL）高层DSL由记录流（KStream）和日志流（KTable/GlobalKTable）两大抽象类组成，包含了一系列调用的实现方法。KStream是对记录流的抽象，其中每条数据都是无限数据集中的一个简单的key-value，KStream提供了处理数据流的多种功能。例如：map、mapValue、flatMap、flatMapValues、filter；它还支持多流连接和流数据的聚合。KTable是对变更日志流的抽象，其中具有相同键的行被覆盖，因此每条数据记录都被视为插入或更新。②ProcessorAPI（低级处理器）低级处理器API扩展抽象类（AbstractProcessor）覆盖包含业务逻辑的处理方法，从而实现对客户端流数据的访问，并允许执行相应的业务逻辑在输入数据流上。同时，其结果作为下游数据转发给客户端。高级DSL以函数式风格提供开箱即用的方法，而低级处理器API则按需提供处理逻辑。③KafkaStreamAPIApplication—AdvancedDSLDemo前提条件：当前环境中必须存在以下依赖，版本视当前情况而定。导入如下包：Kafka配置属性：实例化KStreamBuilder，创建KStream对象：KStreamBuilder有一个Stream方法，以主题名（topicname）为参数，返回一个KStream对象，即订??阅的实例化对象一个指定的话题。基于KStream对象，我们可以使用KafkaStreams高级DSL提供的很多方法（例如：map、process、transform、join等），然后将处理后的数据发送到另一个topic。最后，通过构建器和流配置进行流式传输：使用KafkaStreamsAPI，我们可以在Kafka中流式传输数据，而无需部署单独的集群。KafkaStreamsAPI给我们带来的便利主要包括以下几个方面：高扩展性、灵活性、分布式和容错性。支持有状态和无状态处理。使用窗口、连接和聚合的事件时间处理。通过KafkaStreamsDSL或较低级别的处理器API使用已定义的通用转换操作。没有单独的集群处理要求（与Kafka集成）。一次处理一条记录以实现毫秒级处理延迟。支持KafkaConnect连接不同的应用和数据库。总结Kafka便捷的操作是其受到业界广泛关注的原因之一，但更重要的是它具有出色的稳定性、可靠性和持久性，并且具有灵活的release/queue，可以很好地适应N个Consumer群体，具有强大的可复制性，可以为生产者提供一致性保证。本次分享基于Kafka的核心要素及其通用部署。做了详细的分析，希望能给圈内有兴趣的朋友提供技术普及和交流互补。作者：罗小罗简介：英国计算机专业TOP10，计算机科学与技术硕士，曾就职于汇丰银行、摩根大通、惠普、交通银行、阿里等国内外知名企业。涉及的项目领域主要包括：互联网金融、电子商务、教育、医疗等。目前就职于某世界500强企业，作为测试开发团队的负责人，带领团队搭建并持续优化自动化测试框架，并开发自动化测试辅助工具；擅长领域：单元/接口/性能/安全/自动化测试/CD/CI/DevOps；个人持续研究领域：自动化测试模型/数据分析/算法/机器学习等编辑：陶佳龙征稿：有意投稿或寻求报告，请联系editor@51cto.com合作站点转载请注明原作者和出处为.com]
