AI 助理
备案控制台
开发者社区 微服务 文章 正文

玩转Kafka—初步使用

2024-02-22 55
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
注册配置 MSE Nacos/ZooKeeper，118元/月
日志服务 SLS，月写入数据量 50GB 1个月
云原生网关 MSE Higress，422元/月
简介： 玩转Kafka—初步使用

玩转Kafka—初步使用

官方文档：http://kafka.apache.org/

中文文档：https://kafka.apachecn.org/

1 简单介绍

首先是一些概念:

  • Kafka作为一个集群，运行在一台或者多台服务器上.
  • Kafka 通过 topic 对存储的流数据进行分类。
  • 每条记录中包含一个key，一个value和一个timestamp（时间戳）。

Kafka有四个核心的API:

  • The Producer API 允许一个应用程序发布一串流式的数据到一个或者多个Kafka topic。
  • The Consumer API 允许一个应用程序订阅一个或多个 topic ，并且对发布给他们的流式数据进行处理。
  • The Streams API 允许一个应用程序作为一个流处理器，消费一个或者多个topic产生的输入流，然后生产一个输出流到一个或多个topic中去，在输入输出流中进行有效的转换。
  • The Connector API 允许构建并运行可重用的生产者或者消费者，将Kafka topics连接到已存在的应用程序或者数据系统。比如，连接到一个关系型数据库，捕捉表（table）的所有变更内容。

支持的语言(除了Java之外的)：

常见概念：

1 Topics和日志

让我们首先深入了解下Kafka的核心概念:提供一串流式的记录— topic

Topic 就是数据主题，是数据记录发布的地方,可以用来区分业务系统。Kafka中的Topics总是多订阅者模式，一个topic可以拥有一个或者多个消费者来订阅它的数据。

对于每一个topic， Kafka集群都会维持一个分区日志，如下所示：

每个分区都是有序且顺序不可变的记录集，并且不断地追加到结构化的commit log文件。分区中的每一个记录都会分配一个id号来表示顺序，我们称之为offset，offset用来唯一的标识分区中每一条记录。

Kafka 集群保留所有发布的记录—无论他们是否已被消费—并通过一个可配置的参数——保留期限来控制. 举个例子， 如果保留策略设置为2天，一条记录发布后两天内，可以随时被消费，两天过后这条记录会被抛弃并释放磁盘空间。Kafka的性能和数据大小无关，所以长时间存储数据没有什么问题.

日志中的 partition（分区）有以下几个用途。第一，当日志大小超过了单台服务器的限制，允许日志进行扩展。每个单独的分区都必须受限于主机的文件限制，不过一个主题可能有多个分区，因此可以处理无限量的数据。第二，可以作为并行的单元集—关于这一点，更多细节如下

2 分布式

日志的分区partition （分布）在Kafka集群的服务器上。每个服务器在处理数据和请求时，共享这些分区。每一个分区都会在已配置的服务器上进行备份，确保容错性.

每个分区都有一台 server 作为 “leader”，零台或者多台server作为 follwers 。leader server 处理一切对 partition （分区）的读写请求，而follwers只需被动的同步leader上的数据。当leader宕机了，followers 中的一台服务器会自动成为新的 leader。每台 server 都会成为某些分区的 leader 和某些分区的 follower，因此集群的负载是平衡的。

3 生产者

生产者可以将数据发布到所选择的topic中。生产者负责将记录分配到topic的哪一个 partition（分区）中。可以使用循环的方式来简单地实现负载均衡，也可以根据某些语义分区函数(例如：记录中的key)来完成。下面会介绍更多关于分区的使用。

4 消费者

消费者使用一个 消费组 名称来进行标识，发布到topic中的每条记录被分配给订阅消费组中的一个消费者实例.消费者实例可以分布在多个进程中或者多个机器上。

如果所有的消费者实例在同一消费组中，消息记录会负载平衡到每一个消费者实例.

如果所有的消费者实例在不同的消费组中，每条消息记录会广播到所有的消费者进程.

如图，这个 Kafka 集群有两台 server 的，四个分区(p0-p3)和两个消费者组。消费组A有两个消费者，消费组B有四个消费者。

通常情况下，每个 topic 都会有一些消费组，一个消费组对应一个"逻辑订阅者"。一个消费组由许多消费者实例组成，便于扩展和容错。这就是发布和订阅的概念，只不过订阅者是一组消费者而不是单个的进程。

在Kafka中实现消费的方式是将日志中的分区划分到每一个消费者实例上，以便在任何时间，每个实例都是分区唯一的消费者。维护消费组中的消费关系由Kafka协议动态处理。如果新的实例加入组，他们将从组中其他成员处接管一些 partition 分区;如果一个实例消失，拥有的分区将被分发到剩余的实例。

Kafka 只保证分区内的记录是有序的，而不保证主题中不同分区的顺序。每个 partition 分区按照key值排序足以满足大多数应用程序的需求。但如果你需要总记录在所有记录的上面，可使用仅有一个分区的主题来实现，这意味着每个消费者组只有一个消费者进程。

保证

high-level Kafka给予以下保证:

  • 生产者发送到特定topic partition 的消息将按照发送的顺序处理。 也就是说，如果记录M1和记录M2由相同的生产者发送，并先发送M1记录，那么M1的偏移比M2小，并在日志中较早出现
  • 一个消费者实例按照日志中的顺序查看记录.
  • 对于具有N个副本的主题，我们最多容忍N-1个服务器故障，从而保证不会丢失任何提交到日志中的记录.

关于保证的更多细节可以看文档的设计部分。

2 下载安装

Kafka依赖于Zookeeper，而Zookeeper又依赖于Java，因此在使用Kafka之前要安装jdk1.8的环境和启动zookeeper服务器。

下载或安装地址：

好，下面我们开始进行安装

[root@iZ2ze4m2ri7irkf6h6n8zoZ local]# tar -zxf kafka_2.11-1.0.0.tgz
[root@iZ2ze4m2ri7irkf6h6n8zoZ local]# mv kafka_2.11-1.0.0 kafka-2.11

3 基本使用

3.1 启动Kafka

首先检查下自己的jdk 是否安装：

[root@iZ2ze4m2ri7irkf6h6n8zoZ local]# java -version
java version "1.8.0_144"
Java(TM) SE Runtime Environment (build 1.8.0_144-b01)
Java HotSpot(TM) 64-Bit Server VM (build 25.144-b01, mixed mode)

启动Zookeeper：

[root@iZ2ze4m2ri7irkf6h6n8zoZ zookeeper-3.5.9]# ls
bin  conf  docs  lib  LICENSE.txt  NOTICE.txt  README.md  README_packaging.txt
[root@iZ2ze4m2ri7irkf6h6n8zoZ zookeeper-3.5.9]# cd conf/
[root@iZ2ze4m2ri7irkf6h6n8zoZ conf]# ls
configuration.xsl  log4j.properties  zoo_sample.cfg
[root@iZ2ze4m2ri7irkf6h6n8zoZ conf]# cp zoo_sample.cfg zoo.cfg
[root@iZ2ze4m2ri7irkf6h6n8zoZ conf]# cd ../bin/
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ls
README.txt    zkCli.cmd  zkEnv.cmd  zkServer.cmd            zkServer.sh          zkTxnLogToolkit.sh
zkCleanup.sh  zkCli.sh   zkEnv.sh   zkServer-initialize.sh  zkTxnLogToolkit.cmd
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./zkServer.
zkServer.cmd  zkServer.sh   
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper-3.5.9/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

启动Kafka：

[root@iZ2ze4m2ri7irkf6h6n8zoZ kafka-2.11]# ls
bin  config  libs  LICENSE  NOTICE  site-docs
[root@iZ2ze4m2ri7irkf6h6n8zoZ kafka-2.11]# cd config/
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# ls
connect-console-sink.properties    connect-file-source.properties  log4j.properties        zookeeper.properties
connect-console-source.properties  connect-log4j.properties        producer.properties
connect-distributed.properties     connect-standalone.properties   server.properties
connect-file-sink.properties       consumer.properties             tools-log4j.properties
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# cd ../bin/
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server.properties 
[2021-11-20 10:21:10,326] INFO KafkaConfig values: 
......
[2021-11-20 10:21:12,423] INFO Kafka version : 1.0.0 (org.apache.kafka.common.utils.AppInfoParser)
[2021-11-20 10:21:12,423] INFO Kafka commitId : aaa7af6d4a11b29d (org.apache.kafka.common.utils.AppInfoParser)
[2021-11-20 10:21:12,424] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

3.2 简单测试使用

新建和查看topic

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic ymx
Created topic "ymx".
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --list --zookeeper localhost:2181
ymx

生产者发送消息：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic ymx
>Hello Kafka!
>Hello Ymx!
>Hello Kafka and Ymx!
>

消费者消费消息：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ymx --from-beginning
Hello Kafka!
Hello Ymx!
Hello Kafka and Ymx!

3.3 搭建多代理集群

3.3.1 开始搭建

首先要copy下配置文件

[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# cp server.properties server-01.properties 
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# cp server.properties server-02.properties 
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# vim server-01.properties 
####  内容开始  ####
broker.id=1                  #  21行左右，broker的唯一标识（同一个集群中）
listeners=PLAINTEXT://:9093  #  31行左右，放开，代表kafka的端口号
log.dirs=/tmp/kafka-logs-01  #  60行左右，用逗号分隔的目录列表，在其中存储日志文件  
####  内容结束  ####
[root@iZ2ze4m2ri7irkf6h6n8zoZ config]# vim server-02.properties 
####  内容开始  ####
broker.id=2                  #  21行左右，broker的唯一标识（同一个集群中）
listeners=PLAINTEXT://:9094  #  31行左右，放开，代表kafka的端口号
log.dirs=/tmp/kafka-logs-02  #  60行左右，用逗号分隔的目录列表，在其中存储日志文件  
####  内容结束  ####

根据配置文件启动Kafka（同一主机下）

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server-01.properties

报错信息：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server-01.properties 
Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000000c0000000, 1073741824, 0) failed; error='Cannot allocate memory' (errno=12)
#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map 1073741824 bytes for committing reserved memory.
# An error report file with more information is saved as:
# /usr/local/kafka-2.11/bin/hs_err_pid4036.log

原因：物理机或虚拟机内存不足，不足以保证Kafka启动或运行时需要的内容容量

解决方式：

  • 增加物理机或虚拟机的内存
  • 减少Kafka启动所需内容的配置，将要修改的文件为kafka-server-start.sh
export KAFKA_HEAP_OPTS="-Xmx512M -Xms256M" #29行左右
3.3.2 使用

解决好之后我们开始启动：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server-01.properties 
[2021-11-20 10:58:33,138] INFO KafkaConfig values:
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-server-start.sh ../config/server-02.properties 
[2021-11-20 10:59:04,187] INFO KafkaConfig values:

ps:看下我们的阿里云服务器的状况

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic mr-yan
Created topic "mr-yan".
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic mr-yan
Topic:mr-yan    PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: mr-yan   Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2
  • PartitionCount：主题分区数。
  • ReplicationFactor：用来设置主题的副本数。
  • leader：是负责给定分区所有读写操作的节点。每个节点都是随机选择的部分分区的领导者。
  • replicas：是复制分区日志的节点列表，不管这些节点是leader还是仅仅活着。
  • isr：是一组“同步”replicas，是replicas列表的子集，它活着并被指到leader。

进行集群环境下的使用：

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic mr-yan
>Hello Kafkas!  
>Hello Mr.Yan   
>
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic mr-yan
Hello Kafkas!
Hello Mr.Yan
3.3.3 验证容错性

首先我们停掉一个Kafka的Broker：

[root@iZ2ze4m2ri7irkf6h6n8zoZ ~]# ps -ef|grep server-01.properties
root     19859 28247  1 10:58 pts/3    ../config/server-01.properties
root     23934 16569  0 11:12 pts/11   00:00:00 grep --color=auto server-01.properties
[root@iZ2ze4m2ri7irkf6h6n8zoZ ~]# kill -9 28247
[root@iZ2ze4m2ri7irkf6h6n8zoZ ~]# ps -ef|grep server-01.properties
root     32604 16569  0 11:13 pts/11   00:00:00 grep --color=auto server-01.properties
[root@iZ2ze4m2ri7irkf6h6n8zoZ ~]# cd /usr/local/kafka-2.11/bin/
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic mr-yan
Topic:mr-yan    PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: mr-yan   Partition: 0    Leader: 0       Replicas: 1,0,2 Isr: 0,2

查看生产者和消费者的变化，并再次使用，发现仍可以进行使用

[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic mr-yan
>Hello Kafkas!  
>Hello Mr.Yan   
>[2021-11-20 11:12:28,881] WARN [Producer clientId=console-producer] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
>Hello Kafkas too!
>Hello Mr.Yan too!
>
[root@iZ2ze4m2ri7irkf6h6n8zoZ bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic mr-yan
Hello Kafkas!
Hello Mr.Yan
[2021-11-20 11:12:28,812] WARN [Consumer clientId=consumer-1, groupId=console-consumer-22158] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2021-11-20 11:12:29,165] WARN [Consumer clientId=consumer-1, groupId=console-consumer-22158] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
Hello Kafkas too!
Hello Mr.Yan too!

4 小总结

主题，分区，副本的概念

Kafka是根据主题（topic）进行消息的传递，但是又有分区和副本的概念，下面来分别解释下：

  • 分区：kafka对每一条消息的key做一个hashcode运算，然后将得到的数值对分区数量进行模运算就得到了这条消息所在分区的数字。
  • 副本：同一分区的几个副本之间保存的是相同的数据，副本之间的关系是“一主多从”，其中的主（leader）则负责对外提供读写操作的服务，而从（follower）则负责与主节点同步数据，当主节点宕机，从节点之间能重新选举leader进行对外服务。

kafka会保证同一个分区内的消息有序，但是不保证主题内的消息有序。

参考：https://kafka.apachecn.org/quickstart.html

文章标签：
云消息队列 Kafka 版
微服务引擎
日志服务
消息中间件
Kafka
API
Java
存储
海风极客
目录
相关文章
1176112968452250
|
3月前
|
消息中间件 缓存 算法
kafka(三)
kafka(三)
1176112968452250
38 5
1176112968452250
|
3月前
|
消息中间件 存储 缓存
kafka(一)
kafka(一)
1176112968452250
53 3
yuanzhengme
|
4月前
|
消息中间件 Kafka
kafka里的acks是什么
【8月更文挑战第3天】kafka里的acks是什么
yuanzhengme
226 0
孙玉洁-47170
|
7月前
|
消息中间件 分布式计算 Java
kafka
kafka
孙玉洁-47170
95 0
Sunny_yiyi
|
7月前
|
消息中间件 Java Kafka
Kafka
Kafka
Sunny_yiyi
54 1
还在路上的秃头
|
7月前
|
消息中间件 存储 分布式计算
kafka 详细介绍
kafka 详细介绍
还在路上的秃头
113 0
waynaqua
|
消息中间件 缓存 算法
Kafka为什么这么快？
Kafka 是一个基于发布-订阅模式的消息系统，它可以在多个生产者和消费者之间传递大量的数据。Kafka 的一个显著特点是它的高吞吐率，即每秒可以处理百万级别的消息。那么 Kafka 是如何实现这样高得性能呢？本文将从七个方面来分析 Kafka 的速度优势。
waynaqua
82 1
阿甘兄
|
消息中间件 开发框架 Java
113 Kafka介绍
113 Kafka介绍
阿甘兄
88 0
王二蛋呀
|
消息中间件 缓存 Java
Kafka介绍
Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。 Kafka是一种高吞吐量的分布式发布订阅消息系统，作为消息中间件来说都起到了系统间解耦、异步、削峰等作用，同时又提供了Kafka streaming插件包在应用端实现实时在线流处理，它可以收集并处理用户在网站中的所有动作流数据以及物联网设备的采样信息
王二蛋呀
175 0
预测性维护
|
消息中间件 分布式计算 Java
浅谈kafka 一
浅谈kafka 一
预测性维护
93 0

热门文章

最新文章

  • 1
    四大触点，教你从“用户视角”构建数据分析体系
  • 2
    Spring-boot+Dubbo应用启停源码分析
  • 3
    《Stata统计分析与应用（第2版）》一3.4 图形的保存、合并及修改
  • 4
    RocketMq-Request-Reply消息
  • 5
    Confluence 6 手动添加用户
  • 6
    2013年4月20日 通用权限管理系统组件V3.8功能改进说明
  • 7
    Configuring EIGRP Unequal-Cost Paths
  • 8
    PowerShell 指定关键词遍历文件内容
  • 9
    Windows时间服务出现故障 向用户发送错误的时间
  • 10
    三种dedecms调用相关文章的方法
  • 1
    《鸿蒙HarmonyOS应用开发从入门到精通（第2版）》简介
    54
  • 2
    《Python 助力：人工智能模型的“瘦身”与“加速”之旅》
    111
  • 3
    《非计算机专业者：Python 开启人工智能职业新航道》
    40
  • 4
    《Python赋能：智绘智慧城市新蓝图》
    43
  • 5
    《智育新篇：点亮学生人工智能伦理与社会责任之光》
    47
  • 6
    《数据主权：人工智能时代的核心基石与挑战》
    33
  • 7
    鸿蒙HarmonyOS应用开发 | 探索 HarmonyOS Next-从开发到实战掌握 HarmonyOS Next 的分布式能力
    110
  • 8
    鸿蒙原生开发环境的创新与性能优化：从工具到跨平台支持的全方位提升
    276
  • 9
    鸿蒙HarmonyOS应用开发 | 「鸿蒙技术分享」HarmonyOS NEXT元服务卡片实战体验
    163
  • 10
    Qwen-Coder：通过Qwen 2.5模型实现智能代码生成的技术实践
    41

    • 相关课程

    更多
  • 分布式消息系统 Kafka 快速入门
  • 消息队列Kafka入门课程
  • RocketMQ 5.0学习资料
  • 带您快速了解阿里云消息队列RocketMQ 5.0版
  • 消息队列 RocketMQ 消息集成
  • 分布式协调系统 Zookeeper 快速入门

    • 相关电子书

    更多
  • 消息队列kafka介绍
  • Apache Kafka-and the Rise of
  • 消息队列 Kafka 版差异化特性

    • 相关实验场景

    更多
  • RocketMQ中使用Java客户端发送消息和消费的应用
    • 下一篇
    DataWorks售前咨询