KAFKA

能够无缝处理多个生产者

Kafka 专为多个消费者设计，无需相互干扰即可读取任何单个消息流

消息代理用于将数据处理与数据生产者分离。

使用 Kafka 可以将日志或事件数据抽象为消息流，从而消除对文件细节的任何依赖。

Topics 就像一个类别/一个索引，它把消息一起组在一起

Kafka 中的消息被分类为 Topics。

Topics 还被分解为多个 Partitions。回到"提交日志"描述，分区是单个日志

将消息推送到 Kafka Topics 的流程

Producer 创建新消息

使用来自 Kafka Topics 的消息的进程

消费者阅读信息。

消费者通过跟踪消息的偏移量来 跟踪它已经消费了哪些消息

一个不可变的 Topics 消息序列，连续地附加到结构化提交日志中

分区也是 Kafka 提供冗余性和可扩展性的方式

由于 Topics 通常具有多个分区，因此不能保证整个 Topics（只需在单个分区内）的消息时间排序

单个 Kafka 服务器称为一个broker

代理接收来自生产者的消息，为其分配偏移量，并将消息提交到 磁盘上的存储

它还为使用者提供服务，响应分区的提取请求，并响应已提交到磁盘的消息

消费者作为消费者群体的一部分工作，消费者群体是一个或多个消费者共同消费一个主题。

该组确保每个分区仅由一个成员使用。

消费者可以横向扩展以使用包含大量消息的主题

如果单个使用者 失败，组的其余成员将重新平衡正在使用的分区，以接管丢失的成员。

Kafka 内部的数据单位称为消息

可以把消息本身理解成一个不透明的数组

消息被成批写入 Kafka。批处理只是消息的集合，所有这些消息都是针对同一 Topics 和分区生成的。

Apache Kafka 使用 Zookeeper 存储有关 Kafka 群集的元数据以及使用者客户端详细信 息

Kafka 代理接收来自生产者的消息，并将它们存储在由唯一偏移键键的磁盘上

Kafka 代理允许使用者按 topic、分区和偏移提取消息

Kafka 代理可以通过使用 Zookeeper 直接或间 接地相互共享信息来创建 Kafka 群集

Kafka 群集只有一个代理充当控制器

创建并验证 topic

测试主题生成消息

测试使用来自主题的消息

每个 Kafka 代理都必须有一个整数标识符，该标识符是用来设置broker.id 配置。

默认情况下，此整数设置为 0，但可以是任何值。但在一个 Kafka  cluster中必须是唯一的。

示例配置文件使用 TCP 端口 9092 上的侦听器启动 Kafka。这可以通过更改端口 配置参数设置为任何可用端口

用于存储代理元数据的 Zookeeper 的位置。

hostname、主机名或 Zookeeper 服务器的 IP 地址

port，服务器的客户端端口号

/path，可选的 Zookeeper 路径，用作卡夫卡的 chroot 环境

Kafka 将所有消息保存到磁盘，这些日志段存储在日志目录配置

Kafka 服务器配置为创建的主题指定了许多默认配置。

num.partitions

log.retention.ms

log.retention.bytes

log.segment.bytes

创建 Kafka 主题

Describe a topic

Topics 列表

更改 Topic

此工具用于将消息写入 topic

当消息采用基于文本的格式时，它很有用

向 topic 发送简单的字符串消息

用 key 发送信息

从文件发送消息

从 Kafka 主题读取数据并写入标准输出（控制台）

消费消息

consume 旧message

显示 key-value 信息

kafka-simple-consumer-shell

列出消费者组

删除消费者组

Kafka 主题被划分为多个分区

分区允许您通过在多个代理之间拆分特定主题中的数据来并行化主题。
每个分区都可以放在单独的机器上，以允许多个使用者并行地从一个主题中读取数据。

将对象转换为字节流以进行传输的过程称为序列化

broker是一个无状态的 Kafka 服务器

Kafka 生产者和消费者不直接交互，而是使用 Kafka 服务器作为代理来交换消息服务。

public CreateTopicsResult createTopics(Collection<NewTopic> newTopics)

public DeleteTopicsResult deleteTopics(Collection<String> topics)

public ListTopicsResult listTopics()

public DescribeTopicsResult describeTopics(Collection<String> topicNames)

key.serializer class:实现的键的序列化程序类

value.serializer class： 实现的值的序列化程序类

acks string：生产者要求领导者在考虑完成请求之前收到的确认数。这将控制发送的记录的持
久性。

acks=0 如果设置为零，则生产者将不等待来自服务器的任何确认

acks=1 这意味着领导者会将记录写入本地日志，但不会等待所有追随者的完全确认。

acks=all 这意味着领导者将等待所以的同步副本集来确认记录。这可确保只要至少一 个同步副本保持活动状态，记录就会丢失。这是最强的保证。这等效于 acks=-1 设 置。

bootstrap.servers list

buffer.memory

max.block.ms long

max.request.size int

request.timeout.ms int

同步发送消息

异步发送消息

Properties props = new Properties(); props.setProperty("bootstrap.servers", "localhost：9092"); props.put("acks", "all"); props.put("key.serializer",  "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer",  "org.apache.kafka.common.serialization.StringSerializer");

Properties props = new Properties();  props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost：9092");  props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,  LongSerializer.class.getName());  props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,  StringSerializer.class.getName());

Producer<String, String> producer = new KafkaProducer<>(props);

ProducerRecord<Integer, String> record = new ProducerRecord<String,  String>("topicName", “key1” “value1”);

producer.send(record);

producer.close();

<artifactId>kafka-clients</artifactId>

每个 topic 每个分区的偏移量管理（自动读取 Zookeeper 中消费者组的最后一个
偏移量）

代理故障转移，以及添加或减去分区和消费者时

负载平衡

整合 API

减少依赖性

更好的安全性

更好地控制 Kafka 消息的过度消耗

更灵活的控制

KafkaConsumer<String, String> consumer 
= new KafkaConsumer<>( props );

subcribe()方法将 topic 列表作为参数，以订阅: consumer.subscribe(Collections.singletonList("topicName"));

一旦消费者订阅了 topic，poll 循环将处理协调、分区重新平衡、心跳和数据
获取的所有细节

Properties props = new Properties(); 
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER_URL + ":" + 
KAFKA_SERVER_PORT); 
props.put(ConsumerConfig.GROUP_ID_CONFIG, CLIENT_ID); 
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); 
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); 
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); 
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
"org.apache.kafka.common.serialization.IntegerDeserializer"); 
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
"org.apache.kafka.common.serialization.StringDeserializer);

KafkaConsumer<Integer,String> consumer = new KafkaConsumer<>(props);

consumer.subscribe(Collections.singletonList(this.topic));

ConsumerRecords<Integer,String> records = consumer.poll(100)

for (ConsumerRecord<Integer, String>record : records) {  System.out.println("Received message:  (" + record.key() + ", + record.value() + ") at offset  " + record.offset());  }

bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic my-topic --
replication-factor 1 --partitions 1

kafka-topics.sh --bootstrap-server localhost:2181 --alter --topic mytopic --
partitions 2

目前的版本 Kafka 不允许减少分区的操作, 如果那样做会导致 InvalidPartitionsException. 异常发生

> bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic my-topic

如果想要真正将其删除，需要在 server.properties 里面加上一行配置如下：
delete.topic.enable=true

> bin/kafka-topics.sh --zookeeper localhost:2181 --list

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

可以通过将--list 更改为--describe 并添加--group 参数
来获得更多详情信息，比如偏移量。这将列出
指定消费者组正在使用的所有主题，
以及每个主题分区的偏移量

> bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 -delete --group 
my-group --group my-other-group

当要删除的消费者组不为空时，执行上面命令你将得到以下错误: GroupNotEmptyException。

手动删除一个或多个消费者组，可以使用--delete 

使用--delete-offsets 选项删除消费者组的偏移
量此选项同时支持一个消费者组和一个或多个主题

重置使用者组的偏移量，可以使用“--reset-offsets”选项。此选项一次只支持一个消费者组。它需
要定义以下范围:--all-topic 或--topic

Kafka 客户端唯一可以覆盖的配置是生产者和消费者的配额，即允许具有指定客户端 ID 的所有客户
端在每个 broker 上每秒生产或者消费的字节数

可以使用命令行工具 kafka-configs.sh 来检查主题或客户机的特定配置。显示覆盖的配置
(Describing Configuration Overrides)需要使用--describe 选项。

可以完全删除动态配置，这将导致集群配置恢复到默认值，要删除配置覆盖，请使用--alter 命令以及
--delete-config 命令。

kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics -- entity-name my-topic --delete-config retention.ms

事件时间

日志附加时间

处理时间

本地或内部状态

外部状态

流包含更改的历史记录

表包含当前状态

窗口大小

窗口移动的频率（提前间隔）

窗口保持可更新的时间

流处理的最基本模式是单独地处理每个事件。这也被称为 map/filter 模式

它通常用于从流中过滤不必要的事件或转换每个事件

在这种模式中，流处理应用程序使用流中的事件，修改每个事件，然后
将事件生成到另一个流。

例如，一个应用程序从流中读取日志消息并将错误 事件写入高优先级流，将其余事件写入低优先级流。另一个例子是从流中读 取事件并将其从 JSON 修改为 Avro 的应用程序

大多数流处理应用程序都关注聚合信息，尤其是时间窗口聚合

每个 streams 应用程序至少实现和执行一个拓扑

Kafka Streams 通过允许在应用程序的一个实例中执行多个线程以及支持应
用程序的分布式实例之间的负载平衡来进行扩展

Streams 引擎通过将拓扑拆分为任务来并行执行拓扑

允许我们扩展应用程序的同一个模型也允许我们处理失败