KafkaStreams1、概述Kafka一直被认为是一个强大的消息中间件，它实现了高吞吐、高可用和低延时的消息传输能力，这让它成为流式处理系统中完美的数据来源。目前通用的一些流式处理框架如ApacheSpark、ApacheFlink、ApacheStorm等都可以将Kafka作为可靠的数据来源。但遗憾的是，在0.l0.x版本之前，Kafka还并不具备任何数据处理的能力，但在此之后，KafkaStreams应运而生。KafkaStreams是一个用于处理和分析数据的客户端库。它先把存储在Kafka中的数据进行处理和分析，然后将最终所得的数据结果回写到Kafka或发送到外部系统。它建立在一些非
SpringBoot集成Kafka1、构建项目1.1、引入依赖<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.2.5.RELEASE</version></parent><dependencies><dependency><groupId>org.springframework.boot&
__consumer_offsets详解一般情况下，当集群中第一次有消费者消费消息时会自动创建主题consumeroffsets,不过它的副本因子还受offsets..topic.replication.factor参数的约束，这个参数的默认值为3（下载安装的包中此值可能为l)，分区数可以通过offsets.topic.num.partitions参数设置，默认为50。客户端提交消费位移是使用OffsetCommitRequest请求实现的，OffsetCommitRequest的结构如图所示。请求体第一层中的group_id、generation_id和member_id在前面的内容中已经介
再均衡原理新版的消费者客户端对此进行了重新设计，将全部消费组分成多个子集，每个消费组的子集在服务端对应一个GroupCoordinator对其进行管理，GroupCoordinator是Kafka服务端中用于管理消费组的组件。而消费者客户端中的ConsumerCoordinator组件负责与GroupCoordinator进行交互。ConsumerCoordinator与GroupCoordinator之间最重要的职责就是负责执行消费者再均衡的操作，包括前面提及的分区分配的工作也是在再均衡期间完成的。就目前而言，一共有如下几种情形会触发再均衡的操作：有新的消费者加入消费组。有消费者宕机下线。消
时间轮Kafka中存在大量的延时操作，比如延时生产、延时拉取和延时删除等。Kafka并没有使用JDK自带的Timer或DelayQueue来实现延时的功能，而是基于时间轮的概念自定义实现了一个用于延时功能的定时器(SystemTimer)。JDK中Timer和DelayQueue的插入和删除操作的平均时间复杂度为O(nlogn)并不能满足Kafka的高性能要求，而基于时间轮可以将插入和删除操作的时间复杂度都降为O(I)。时间轮的应用并非Kafka独有，其应用场景还有很多，在Netty、Akka、Quartz、ZooKeeper等组件中都存在时间轮的踪影。如下图所示，Kafka中的时间轮(Tim
指定位移消费试想一下，当一个新的消费组建立的时候，它根本没有可以查找的消费位移。或者消费组内的一个新消费者订阅了一个新的主题，它也没有可以查找的消费位移。当consumeroffsets主题中有关这个消费组的位移信息过期而被删除后，它也没有可以查找的消费位移。在Kafka中每当消费者查找不到所记录的消费位移时，就会根据消费者客户端参数auto.offset.reset的配置来决定从何处开始进行消费，这个参数的默认值为“latest”,表示从分区末尾开始消费消息。参考下图，按照默认的配置，消费者会从9开始进行消费(9是下一条要写入消息的位置)，更加确切地说是从9开始拉取消息。如果将auto.of
位移提交对于Kafka中的分区而言，它的每条消息都有唯一的offset,用来表示消息在分区中对应的位置。对于消费者而言，它也有一个offset的概念，消费者使用offset来表示消费到分区中某个消息所在的位置。单词“offset”可以翻译为“偏移量”，也可以翻译为“位移”。在每次调用poll()方法时，它返回的是还没有被消费过的消息集（当然这个前提是消息已经存储在Kafka中了，并且暂不考虑异常情况的发生)，要做到这一点，就需要记录上一次消费时的消费位移。并且这个消费位移必须做持久化保存，而不是单单保存在内存中，否则消费者重启之后就无法知晓之前的消费位移。再考虑一种情况，当有新的消费者加入时，
KafkaConnect1、概要介绍KafkaConnect是一个高伸缩性、高可靠性的数据集成工具，用于在ApacheKafka与其他系统间进行数据搬运以及执行ETL操作，比如KafkaConnect能够将文件系统中某些文件的内容全部灌入Kafkatopic中或者是把Kafkatopic中的消息导出到外部的数据库系统，如图所示。如图所示，KafkaConnect主要由sourceconnector和sinkconnector组成。事实上，几乎大部分的ETL框架都是由这两大类逻辑组件组成的，如ApacheFlume、Kettle等。sourceconnector负责把输入数据从外部系统中导入到K
MirrorMaker1、概要介绍对于Kafka企业级用户而言，一个常见的痛点就是跨机房或跨数据中心(datacenter,DC)的数据传输。大型企业通常在多个数据中心部署Kafka集群。.这里的数据中心可能是企业拥有的自建机房，也可能是公有云厂商的不同机房。在多个机房部署Kafk集群的优势如下。实现灾备。较近的地理位置可缩短延时以及用户响应时间。实现负载均衡，即每个数据中心上的集群可能只保存部分数据集合。区别隔离不同优先级的数据处理。尽管部署方式各种各样，但跨机房部署方案有一个共同的特点：数据需要能够从一个Kafka集群被拷贝到另一个集群，而且还必须支持双向拷贝：某次传输的源集群(sourc
AdminClient自0.ll.0.0版本起，Kafka社区推出了AdminClient和KafkaAdminClient,意在统一所有的集群管理API。使用0.11.0.0及以后版本的用户应该始终使用这个类来管理集群。虽然和原先服务器端的AdminClient类同名，但这个工具是属于客户端的，因此只需要在管理程序项目中添加kafkaclients依赖即可:<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><
consumer端设计1、consumergroup状态机如前所述，新版本consumer依赖于broker端的组协调者(coordinator)来管理组内的所有consumer实例并负责把分配方案下发到每个consumer上。分配方案是由组内的leaderconsumer根据指定的分区分配策略指定的。该分配策略必须是组内所有consumer都支持的。事实上，如果所有consumer协调在一起无法选出共同的分区策略，那么coordinator就会抛出异常。这种设计就确保了每个consumergroup总有一个一致性的分配策略，同时还能确保每个consumer只能为它拥有的分区提交位移。Kafk
producer端设计1、producer端基本数据结构新版本客户端（包含新版本producer和新版本consumer)重写了之前服务器端代码提供的很多数据结构，以摆脱对服务器端代码的依赖。其中有一些是理解新版本producer设计所必需的，它们包括（但不限于）如下这些。1.1、ProducerRecord一个ProducerRecord封装了一条待发送的消息（或称为记录)。虽然Kafka0.1l.0.0版本对消息格式进行了部分重构以支持事务和精确一次处理语义，但将以0.10.2.1版本的消息格式进行producer的说明。ProducerRecord由5个字段构成，它们分别如下。topic
Controller设计1、Controller概览在一个Kafka集群中，某个broker会被选举出来承担特殊的角色，即控制器（下称controller)。顾名思义，引入controller就是用来管理和协调Kafka集群的。具体来说，就是管理集群中所有分区的状态并执行相应的管理操作。每个Kafka集群任意时刻都只能有一个controller。当集群启动时，所有broker都会参与controller的竞选，但最终只能由一个broker胜出。一旦controller在某个时刻崩溃，集群中剩余的broker会立刻得到通知，然后开启新一轮的controller选举。新选举出来的controlle
副本与ISR设计一个Kafka分区本质上就是一个备份日志，即利用多份相同的备份共同提供冗余机制来保持系统高可用性。这些备份在Kafka中被称为副本(replica)。Kafka把分区的所有副本均匀地分配到所有broker上，并从这些副本中挑选一个作为leader副本对外提供服务，而其他副本被称为follower副本，只能被动地向leader副本请求数据，从而保持与leader副本的同步。假如leader副本永远工作正常，那么其实不需要follower副本。但现实总是残酷的，Kafkaleader副本所在的broker可能因为各种各样的原因而随时宕机。一旦发生这种情况，follower副本会竞相
多线程消费实例，KafkaConsumer是非线程安全的。它和KafkaProducer不同，后者是线程安全的，因此用户可以在多个线程中放心地使用同一个KafkaProducer实例，事实上这也是社区推荐的producer使用方法，因为通常它比每个线程维护一个KafkaProducer实例效率要高。但是对于consumer而言，用户无法直接在多个线程中共享一个KafkaConsumer实例，下面给出两种多线程消费的方法以及各自的实例。1、每个线程维护一个KafkaConsumer在这个方法中，用户创建多个线程来消费topic数据。每个线程都会创建专属于该线程的KafkaConsumer实例，如
重平衡1、rebalance概览consumergroup的rebalance本质上是一组协议，它规定了一个consumergroup是如何达成一致来分配订阅topic的所有分区的。假设某个组下有20个consumer实例，该组订阅了一个有着l00个分区的topic。正常情况下，Kafka会为每个consumer平均分配5个分区。这个分配过程就被称为rebalance。当consumer成功地执行rebalance后，组订阅topic的每个分区只会分配给组内的一个consumer实例。和旧版本consumer依托于ZooKeeper进行rebalance不同，新版本consumer使用了Kaf
无消息丢失配置Java版本producer用户采用异步发送机制。KafkaProducer.send方法仅仅把消息放入缓冲区中，由一个专属I/O线程负责从缓冲区中提取消息并封装进消息batch中，然后发送出去。显然，这个过程中存在着数据丢失的窗口：若I/O线程发送之前producer崩溃，则存储缓冲区中的消息全部丢失了。这是producer需要处理的很重要的问题。producer的另一个问题就是消息的乱序。假设客户端依次执行下面的语句发送两条消息到相同的分区：producer.send(record1);producer.send(record2);若此时由于某些原因（比如瞬时的网络抖动）导致
producer拦截器producer拦截器(interceptor)是一个相当新的功能，它和consumer端interceptor是在Kafka0.l0.0.0版本中被引入的，主要用于实现clients端的定制化控制逻辑。对于producer而言，interceptor使得用户在消息发送前以及producer回调逻辑前有机会对消息做一些定制化需求，比如修改消息等。同时，producer允许用户指定多个interceptor按序作用于同一条消息从而形成一个拦截链(interceptorchain)。intercetpor的实现接口是org.apache.kafka..clients.prod
1、简介Kafka是一个能够支持高并发以及流式消息处理的消息中间件，并且Kafka天生就是支持集群的，今天就主要来介绍一下如何搭建Kafka集群。Kafka目前支持使用Zookeeper模式搭建集群以及KRaft模式（即无Zookeeper）模式这两种模式搭建集群，这两种模式各有各的好处，今天就来分别介绍一下这两种方式1.1、Kafka集群中的节点类型一个Kafka集群是由下列几种类型的节点构成的，它们充当着不同的作用：Broker节点：即代理节点，是Kafka中的工作节点，充当消息队列的角色，负责储存和处理消息，每个Broker都是一个独立的Kafka服务器，可以在不同的机器上运行，除此之外
1、流式处理概述数据流(也被称为“事件流”或“流数据”)。首先，数据流是无边界数据集的抽象表示。无边界意味着无限和持续增长。无边界数据集之所以是无限的，是因为随着时间的推移，新的记录会不断加入进来。这个简单的模型(事件流)可以表示很多业务活动，比如信用卡交易、股票交易、包裹递送、流经交换机的网络事件、制造商设备传感器发出的事件、发送出去的邮件、游戏里物体的移动，等等。这个清单是无穷无尽的，因为几乎每一件事情都可以被看成事件的序列。除了没有边界外，事件流模型还有其他一些属性。事件流是有序的事件的发生总是有个先后顺序。以金融活动事件为例，先将钱存进账户后再花钱，这与先花钱再还钱的次序是完全不一样的
管理KafkaKafka提供了一些命令行工具，用于管理集群的变更。这些工具使用Java类实现，Kafka提供了一些脚本来调用这些Java类。不过，它们只提供了一些基本的功能，无法完成那些复杂的操作。虽然Kafka实现了操作主题的认证和授权控制，但还不支持集群的其他大部分操作。也就是说，在没有认证的情况下也可以使用这些命令行工具，在没有安全检查和审计的情况下也可以执行诸如主题变更之类的操作。1、主题操作（kafka-topic.sh）使用kafka-topics.sh工具可以执行主题的大部分操作(配置变更部分已经被弃用并被移动到kafka-configs.sh工具当中)。我们可以用它创建、修改、
1、可靠的保证Kafka作出的保证：Kafka可以保证分区消息的顺序。如果使用同一个生产者往同一个分区写入消息，而且消息B在消息A之后写入，那么Kafka可以保证消息B的偏移量比消息A的偏移量大，而且消费者会先读取消息A再读取消息B。只有当消息被写入分区的所有同步副本时(但不一定要写入磁盘)，它才被认为是“已提交”的。生产者可以选择接收不同类型的确认，比如在消息被完全提交时的确认，或者在消息被写入首领副本时的确认，或者在消息被发送到网络时的确认。只要还有一个副本是活跃的，那么已经提交的消息就不会丢失。消费者只能读取已经提交的消息。这些基本的保证机制可以用来构建可靠的系统，但仅仅依赖它们是无法保
1、集群成员关系Kafka使用Zookeeper来维护集群成员的信息。每个broker都有一个唯一标识符，这个标识符可以在配置文件里指定，也可以自动生成。在broker启动的时候，它通过创建临时节点把自己的ID注册到Zookeeper。Kafka组件订阅Zookeeper的/brokers/ids路径(broker在Zookeeper上的注册路径)，当有broker加入集群或退出集群时，这些组件就可以获得通知。如果你要启动另一个具有相同ID的broker，会得到一个错误：新broker会试着进行注册，但不会成功，因为Zookeeper里已经有一个具有相同ID的broker。在broker停机、
1、KafkaConsumer概念1.1、消费者和消费者群组Kafka消费者从属于消费者群组。一个群组里的消费者订阅的是同一个主题，每个消费者接收主题一部分分区的消息。假设主题T1有4个分区，我们创建了消费者C1，它是群组G1里唯一的消费者，我们用它订阅主题T1。消费者C1将收到主题T1全部4个分区的消息，如图所示。如果在群组G1里新增一个消费者C2，那么每个消费者将分别从两个分区接收消息。我们假设消费者C1接收分区0和分区2的消息，消费者C2接收分区1和分区3的消息，如图所示。如果群组G1有4个消费者，那么每个消费者可以分配到一个分区，如图所示。如果我们往群组里添加更多的消费者，超过主题的分
1、生产者概览生产者向Kafka发送消息的主要步骤：从创建一个ProducerRecord对象开始，ProducerRecord对象需要包含目标主题和要发送的内容。我们还可以指定键或分区。在发送ProducerRecord对象时，生产者要先把键和值对象序列化成字节数组，这样它们才能够在网络上传输。接下来，数据被传给分区器。如果之前在ProducerRecord对象里指定了分区，那么分区器就不会再做任何事情，直接把指定的分区返回。如果没有指定分区，那么分区器会根据ProducerRecord对象的键来选择一个分区。选好分区以后，生产者就知道该往哪个主题和分区发送这条记录了。紧接着，这条记录被添加
1、简介Kafka包括五个核心apis：ProducerAPI：ProducerAPI用于将流数据发送至一个或多个Kafka主题中。ConsumerAPI：ConsumerAPI用于应用程序在订阅一个或多个主题后，对他们所产生的数据流进行处理。StreamsAPI：StreamsAPI使用来自一个或多个主题的输入流，并生成到一个或多个输出主题的输出流，提供一种有效的数据流处理方式。ConnectorAPI：ConnectorAPI允许构建和运行可重用的生产者或消费者，将Kafka主题连接到现有的应用程序或数据系统。例如，创建一个到关系型数据库连接器,用于捕获对某张表的所有数据变更.AdminC
1、server.properties属性默认值描述broker.id0每个broker都需要有一个标识符，使用broker.id来表示。它的默认值是0，也可以被设置成其他任意整数。这个值在整个Kafka集群里必须是唯一的。这个值可以任意选定，如果出于维护的需要，可以在服务器节点间交换使用这些D。建议把它们设置成与机器名具有相关性的整数，这样在进行维护时，将①号映射到机器名就没那么麻烦了。例如，如果机器名包含唯一性的数字（比如host1.example…com、host2.example.com),那么用这些数字来设置broker.id就再好不过了。log.dirs/tmp/kafka-log
1、下载Kafka安装包下载地址#将压缩包移动到/usr/local/mvkafka_2.12-3.1.0.tgz/usr/local/#解压tar-zxvfkafka_2.12-3.1.0.tgz2、启动启动zookeeper#启动独立安装的zookeeper./zkServer.shstart#也可以自动kafka自带的zookerper./zookeeper-server-start.sh../config/zookeeper.properties启动kafkacd/usr/local/kafka_2.12-3.1.0/bin./kafka-server-start.sh-daemon.
1、简介Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。2、流式数据平台作为一个流式数据平台，最重要的是要具备下