Kafka 使用 | Arterning Blog

😀

Kafka 是一个分布式、高吞吐量、高扩展性的消息队列系统，主要应用在日志收集系统和消息系统。

Kafka 是 LinkedIn 开源的分布式消息订阅系统，目前归属 Apache 顶级开源项目，主要特点是基于 Pull 模式来处理消息消费，追求高吞吐量，一开始用于日志的收集和传输，适合大数据的数据收集业务

Kafka 专为高容量发布/订阅消息和流而设计，旨在持久、快速和可扩展。从本质上讲，Kafka 提供了一个持久的消息存储，类似于日志，其具备的特点如表1所示。

安装 kafka

1.kafka下载地址：http://kafka.apache.org/downloads

解压：kafka_2.12-2.6.0.tgz

2.配置zookeeper

进入config目录找到文件zookeeper.properties，修改dataDir

bash


# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
# 
#    http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# the directory where the snapshot is stored.
dataDir=D:/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=0
# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
admin.enableServer=false
# admin.serverPort=8080

3.配置kafka ，修改日志路径为本地路径

进入config目录找到文件server.properties，修改log.dirs

bash


############################# Log Basics #############################

# A comma separated list of directories under which to store log files
log.dirs=D:/tmp/kafka-logs

4.启动zookeep

bash


D:\kafka_2.12-2.6.0\bin\windows>zookeeper-server-start.bat ..\..\config\zookeeper.properties

5.启动kafka

bash


D:\kafka_2.12-2.6.0\bin\windows>kafka-server-start.bat ..\..\config\server.properties

命令行操作

创建Topic

bash


kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

查看Topic

bash


kafka-topics.bat --list --bootstrap-server localhost:9092

启动生产者 producer

bash


kafka-console-producer.bat --broker-list localhost:9092 --topic test

启动后会接受命令行收入的消息，输入要发送的消息然后回车

可以在GUI中看到接受到的消息

启动消费者 customer

plain


kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning

使用GUI工具：Kafka Tool

Kafka Tool是一个用于管理和使用Apache Kafka集群的GUI应用程序。Kafka Tool提供了一个较为直观的UI可让用户快速查看Kafka集群中的对象以及存储在topic中的消息，提供了一些专门面向开发人员和管理员的功能。

NOTE：目前Kafka Tool已改名为Offset Explorer，不过我还是倾向于叫它 Kafka Tool。

下载地址：https://www.kafkatool.com/download.html

使用方式比较简单，首先创建一个Connection，需要填写的配置如下：

新建Topic

然后更改一下显示内容的类型：从Byte Array改为String，方便查看。

也可以修改全局设置

在Tools -->Settings --> Topics，将 Key 和 Message 均设置为String 保存。这样是全局设置，对所有 Topic 均生效。

可以在Data页签查看Topic中的消息

可以发现，消息在消费后，依然会保存在Topic中，不会删除。Kafka的不同之处在于它是一个非常好的存储系统。写入Kafka的数据将写入磁盘并进行复制以实现容错。

对于传统的消息队列，一般消费过的消息会被删除，而在 Kafka 中消费过的消息不会被删除，始终保留所有的消息，只记录一个消费者消费消息的偏移量（offset，用于记录消费位置）作为标记。Kafka 允许消费者自己设置这个偏移量，允许消费者重复消费一些消息。但始终不删除消费过的消息，日积月累，消息势必会越来越多，占用空间也越来越大。

Kafka 提供了两种策略来删除消息：一种是基于时间，另一种是基于分区文件的大小，我们可以通过配置来决定使用哪种方式。

Kafka 基本概念

Kafka 作为一个高度可扩展可容错的消息系统，它有很多基本概念，下面就来认识一下这些 Kafka 专属的概念

topic

Topic 被称为主题，在 kafka 中，使用一个类别属性来划分消息的所属类，划分消息的这个类称为 topic。topic 相当于消息的分配标签，是一个逻辑概念。主题好比是数据库的表，或者文件系统中的文件夹。

partition

partition 译为分区，topic 中的消息被分割为一个或多个的 partition，它是一个物理概念，对应到系统上的就是一个或若干个目录，一个分区就是一个 提交日志。消息以追加的形式写入分区，先后以顺序的方式读取。

生产者发送的消息主题会被存储在分区中，Kafka 把数据分成多个块，让消息合理地分布在不同的分区，分区被分在不同的 Kafka 实例也就是服务器上，这样就实现了大量消息的负载均衡。

注意：由于一个主题包含无数个分区，因此无法保证在整个 topic 中有序，但是单个 Partition 分区可以保证有序。消息被迫加写入每个分区的尾部。Kafka 通过分区来实现数据冗余和伸缩性

分区可以分布在不同的服务器上，也就是说，一个主题可以跨越多个服务器，以此来提供比单个服务器更强大的性能。

举个不太恰当的类比：可以将topic类比为Mysql中的Table，Partions相当于分表

消费者组（Consumer Group）和消费者（Consumer）

消费者读取 Kafka 中的消息，可以消费任何主题的数据。多个消费者组成一个消费者组，一般消费者必须有一个组（Group）名，如果没有的话会被分一个默认的组名。

一个组可以有多个消费者，一条消息在一个组中，只会被一个消费者获取。

kafka使用场景

1 消息Messaging

Kafka可以替代更传统的消息代理。消息代理的使用有多种原因（将处理与数据生成器分离，缓冲未处理的消息等）。与大多数消息传递系统相比，Kafka具有更好的吞吐量，内置分区，复制和容错功能，这使其成为大规模消息处理应用程序的理想解决方案。

根据经验，消息传递的使用通常相对较低，但可能需要较低的端到端延迟，并且通常取决于Kafka提供的强大的耐用性保证。

在这个领域，Kafka可与传统的消息传递系统（如ActiveMQ或 RabbitMQ）相媲美。

2 网站活动跟踪

Kafka的原始用例是能够将用户活动跟踪管道重建为一组实时发布 - 订阅源。这意味着站点活动（页面查看，搜索或用户可能采取的其他操作）将发布到中心主题，每个活动类型包含一个主题。这些源可用于订购一系列用例，包括实时处理，实时监控以及加载到Hadoop或离线数据仓库系统以进行脱机处理和报告。

活动跟踪通常非常高，因为为每个用户页面视图生成了许多活动消息。

3 度量Metrics

Kafka通常用于运营监控数据。这涉及从分布式应用程序聚合统计信息以生成操作数据的集中式提要。

4 日志聚合

许多人使用Kafka作为日志聚合解决方案的替代品。日志聚合通常从服务器收集物理日志文件，并将它们放在中央位置（可能是文件服务器或HDFS）进行处理。Kafka抽象出文件的细节，并将日志或事件数据作为消息流更清晰地抽象出来。这允许更低延迟的处理并更容易支持多个数据源和分布式数据消耗。与Scribe或Flume等以日志为中心的系统相比，Kafka提供了同样出色的性能，由于复制而具有更强的耐用性保证，以及更低的端到端延迟。

5 流处理

许多Kafka用户在处理由多个阶段组成的管道时处理数据，其中原始输入数据从Kafka主题中消费，然后聚合，丰富或以其他方式转换为新主题以供进一步消费或后续处理。

例如，用于推荐新闻文章的处理管道可以从RSS订阅源抓取文章内容并将其发布到“文章”主题; 进一步处理可能会对此内容进行规范化或重复数据删除，并将已清理的文章内容发布到新主题; 最终处理阶段可能会尝试向用户推荐此内容。此类处理管道基于各个主题创建实时数据流的图形。从0.10.0.0开始，这是一个轻量级但功能强大的流处理库，名为Kafka Streams 在Apache Kafka中可用于执行如上所述的此类数据处理。除了Kafka Streams之外，其他开源流处理工具包括Apache Storm和 Apache Samza。

使用Kafka需要注意的地方

消费者需要引入重试机制

对于partion需要设置合理的路由规则，避免某个partion消息积压，而某些partion数据过少。可以考虑根据ID取模

消费者需要归档历史数据，防止消费能力降低引起消息挤压。

参考

https://www.bilibili.com/read/cv24831491/?spm_id_from=444.41.0.0

💡

还是那句话，学习任何技术，跟学骑自行车一样，不要一开始只关注它的具体细节是什么。先学着怎么骑，骑着骑着就了解大致的原理，这个时候在去看它的原理，会很轻松。

🗒️Kafka 使用

安装 kafka

bash

bash

bash

bash

命令行操作

bash

bash

bash

plain

使用GUI工具：Kafka Tool

Kafka 基本概念

topic

partition

消费者组（Consumer Group）和消费者（Consumer）

kafka使用场景

1 消息Messaging

2 网站活动跟踪

3 度量Metrics

4 日志聚合

5 流处理

使用Kafka需要注意的地方

参考