AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

【Kafka使用场景/基本概念/初体验/消费模式/消费组的偏移量】

2023-05-19 155
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 【Kafka使用场景/基本概念/初体验/消费模式/消费组的偏移量】

文章目录


Kafka

Kafka的使用场景

Kafka基本概念

Kafka的初体验

创建主题

查看kafka中目前存在的topic

删除主题

发送消息

消费消息

消费之前的消息

消费最新的消息

消费之前多主题的消息

消费模式

单播消费

多播消费

查看消费组名

查看消费组的消费偏移量

查看消费组testGroup和testGroup2的消费偏移量

消费组的消费偏移量参数解释：

发送消息进行理解消费组的消费偏移量

发送一条消息

查看偏移量

客户端的消费组testGroup开始消费

再次查看消费组testGroup和testGroup2的消费偏移量


本文内容：


Kafka


7a426c64880a46c98b87ffaeb114ba1c.png


Kafka的使用场景


日志收集：一个公司可以用Kafka收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。

消息系统：解耦和生产者和消费者、缓存消息等。

用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。

运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。


Kafka基本概念


b568ceb3db3e47619503bef264db6c89.png

Broker：消息中间件处理节点，一个Kafka节点就是一个broker，一个或者多个Broker可以组成一个Kafka集群。

Topic：Kafka根据topic对消息进行归类，发布到Kafka集群的每条消息都需要指定一个topic。

Producer：消息生产者，向Broker发送消息的客户端。 Consumer：消息消费者，从Broker读取消息的客户端。

ConsumerGroup：每个Consumer属于一个特定的Consumer Group，一条消息可以被多个不同的ConsumerGroup消费，但是一个Consumer Group中只能有一个Consumer能够消费该消息。

Partition：物理上的概念，一个topic可以分为多个partition，每个partition内部消息是有序的。


Kafka的初体验


创建主题


创建一个名字为“test”的Topic，这个topic只有一个partition，并且备份因子也设置为1。当producer发布一个消息到某个指定的Topic，这个Topic如果不存在，就自动创建。


/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --create --zookeeper 106.14.132.94:2181 --replication-factor 1 --partitions 1 --topic test


查看kafka中目前存在的topic

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --list --zookeeper 106.14.132.94:2181


列表中有一个__consumer_offsets主题，这个主题不能删除哟


删除主题

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --delete --topic test --zookeeper 106.14.132.94:2181


如果出现This will have no impact if delete.topic.enable is not set to true.


彻底删除topic：

[1. ] 删除Topic，delete.topic.enable=true这里要设置为true，需要在$KAFKA/config/server.properties中配置delete.topic.enble=true

[2. ] 删除log日志

[3. ] 删除ZK中的Topic记录


删除列表中有一个__consumer_offsets主题会出现，Topic __consumer_offsets is a kafka internal topic and is not allowed to be marked for deletion.


__consumer_offsets这个topic是由kafka自动创建的，默认49个，这个topic是不能被删除的！


为什么这里会是这样存储__consumer_offsets的呢？

[1.] 将所有 N Broker 和待分配的 i 个 Partition 排序
[2.] 将第 i 个 Partition 分配到第(i mod n)个 Broker 上


发送消息

kafka自带了一个producer命令客户端，可以从本地文件中读取内容，或者我们也可以以命令行中直接输入内容，并将这些内容以消息的形式发送到kafka集群中。在默认情况下，每一个行会被当做成一个独立的消息。首先我们要运行发布消息的脚本，然后在命令中输入要发送的消息的内容。


/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

this is a news

this is a another news


消费消息


对于consumer，kafka同样也携带了一个命令行客户端，会将获取到内容在命令中进行输出，默认是消费最新的消息。由于我们已经发送了消息了，想要消费之前的消息可以通过–from-beginning参数指定。


消费之前的消息
/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --from-beginning --topic test


消费最新的消息
/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --topic test


通过不同的终端窗口来运行以上的命令，你将会看到在producer终端输入的内容，很快就会在consumer的终端窗口上显示出来。


消费之前多主题的消息

/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --from-beginning --whitelist "test|test2"


消费模式


单播消费


一条消息只能被某一个消费者消费的模式，类似queue模式，只需让所有消费者在同一个消费组里即可。


分别在两个客户端执行如下消费命令，然后往主题里发送消息，结果只有一个客户端能收到消息。

发送消息：/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test


5f7555c0f6a547dd9de75ae4fed7d228.png


客户端1：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test


a7fc015303564a2ea0083c4a0a8847be.png


客户端2：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test


03e341f415ca450ead9dcf709320c317.png


多播消费


一条消息能被多个消费者消费的模式，类似publish-subscribe模式费，针对Kafka同一条消息只能被同一个消费组下的某一个消费者消费的特性，要实现多播只要保证这些消费者属于不同的消费组即可。我们再增加一个消费者，该消费者属于testGroup-2消费组，结果两个客户端都能收到消息。


发送消息：/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test


this is a more news

this is a more new2

this is a more new3


3e35c207d2f742f9bc092dac17158f55.png


客户端1：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test


fa92114d96a1410086f5a6d59114784c.png


客户端2：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test


9780f11a39e442c0b1e19f7c157dfdd3.png


客户端3：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup2 --topic test


a93032b14ceb4f179c69fb948e0cc27b.png


查看消费组名


/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --list


0901f87801394f15a5e327a54f786373.png


查看消费组的消费偏移量


查看消费组testGroup和testGroup2的消费偏移量


/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup



/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup2


44e10384f0b74cb094295fd8fa84b210.png

消费组的消费偏移量参数解释：


  • current-offset：当前消费组的已消费偏移量
  • log-end-offset：主题对应分区消息的结束偏移量(HW)
  • lag：当前消费组未消费的消息数


发送消息进行理解消费组的消费偏移量

1个客户端的消费组是testGroup，一个客户端的消费组是testGroup，他们二个都是同一个test主题。


发送一条消息


/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test


this is a testgroup message


查看偏移量
/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup


11a96841c1884716b1d6cbeeffc374ca.png

当前消费组的已消费偏移量（current-offset）没变，说明当前的消费组没有消费者进行消费，因为没有启动消费者。主题对应分区消息的结束偏移量（log-end-offset）加1了，说明当前分区的消息多了一条。当前消费组未消费的消息数（lag）加1了，说明有1条消息没有被消费。


客户端的消费组testGroup开始消费


/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092  --consumer-property group.id=testGroup --topic test


再次查看消费组testGroup和testGroup2的消费偏移量


/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup



/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup2


30a13f8fba934ff1b9032ac3dfc852e9.png


发现testGroup消费组当前消费组的已消费偏移量（current-offset）加1了，当前消费组未消费的消息数（lag）减1了，说明testGroup已经被消费了。testGroup2消费者还是一样没有发生变化，所以可以推测出，消费者是通过不同的消费组进行消费的，每个消费组互不影响。

总结

以上就是今天要讲的内容，还希望各位读者大大能够在评论区积极参与讨论，给文章提出一些宝贵的意见或者建议📝，合理的内容，我会采纳更新博文，重新分享给大家。


🙏四连 关注🔎点赞👍收藏⭐️留言📝


感谢大家的支持，用心写博文分享给大家，你的支持（🔎点赞👍收藏⭐️留言📝）是对我创作的最大帮助。

🍊微信公众号：南北踏尘

🍊主页地址：java_wxid

🍊社区地址：幕后大佬


给读者大大的话


我本身是一个很普通的程序员，放在人堆里，除了与生俱来的🌹盛世美颜🌹、所剩不多的发量，就剩下180的大高个了。就是我这样的一个人，默默坚持写博文也有好多年了，有句老话说的好，🌕牛逼之前都是傻逼式的坚持🌕。希望自己可以通过大量的作品，时间的积累，个人魅力、运气和时机，可以打造属于自己的🌟技术影响力🌟。同时也希望自己可以成为一个🎄懂技术🎄，🎄懂业务🎄，🎄懂管理🎄的综合型人才，作为项目架构路线的总设计师，掌控全局的🌕团队大脑🌕，技术团队中的🍊绝对核心🍊是我未来几年不断前进的目标。

提示：以下都是资源分享，求个一键三连。

面试资料

福利大放送，🎉欢迎关注🔎点赞👍收藏⭐️留言📝，拜托了🙏，这对我真的很重要。

点击：面试资料

提取码：2021


200套PPT模板

福利大放送，🎉欢迎关注🔎点赞👍收藏⭐️留言📝，拜托了🙏，这对我真的很重要。

点击：200套PPT模板

提取码：2021


提问的智慧

福利大放送，🎉欢迎关注🔎点赞👍收藏⭐️留言📝，拜托了🙏，这对我真的很重要。

点击：提问的智慧

提取码：2021


P5学习路线图


a3e4d31525954c9d8b00f1358c3af259.png

P6学习路线图

6fc87924d83748d78113c3b2a295967d.png


P7学习路线图


321588d2a94a47aeb88ccf450c3d5708.png


P8学习路线图


b6d3bfd0204b4380b84c663f89f05f3b.png


以上四张图详细介绍了作为Java开发工作者所需要具备的知识技能，同学们学废了嘛，有想法系统学习的同学可以私聊我，🎉欢迎关注🔎点赞👍收藏⭐️留言📝。

🍊博主：java_wxid

🍊博主：Java廖志伟

🍊社区：幕后大佬

文章标签：
云消息队列 Kafka 版
Java
消息中间件
分布式计算
监控
Hbase
Kafka
Hadoop
程序员
缓存
存储
关键词：
云消息队列 Kafka 版基本概念
云消息队列 Kafka 版模式
云消息队列 Kafka 版偏移量
云消息队列 Kafka 版使用场景
云消息队列 Kafka 版概念
我是廖志伟
目录
相关文章
软件求生
|
2月前
|
消息中间件 存储 Kafka
Kafka日志处理：深入了解偏移量查找与切分文件
**摘要：** 本文介绍了如何在Kafka中查找偏移量为23的消息，涉及ConcurrentSkipListMap的查询、索引文件的二分查找及日志分段的物理位置搜索。还探讨了Kafka日志分段的切分策略，包括大小、时间、索引大小和偏移量达到特定阈值时的切分条件。理解这些对于优化Kafka的性能和管理日志至关重要。
软件求生
82 2
小王老师呀
|
17天前
|
消息中间件 存储 监控
Kraft模式下Kafka脚本的使用
【9月更文挑战第9天】在Kraft模式下，使用Kafka脚本涉及以下几个关键步骤：启动Zookeeper和Kafka服务、创建主题、发送与消费消息、查看主题列表及描述主题详情。通过指定配置文件与相关参数，如`--replication-factor`和`--partitions`，可以灵活管理主题。此外，确保根据实际需求调整配置文件中的参数，并监控日志以维持最佳性能与及时问题处理。
小王老师呀
43 8
土木林森
|
1月前
|
消息中间件 Java Kafka
Kafka不重复消费的终极秘籍！解锁幂等性、偏移量、去重神器，让你的数据流稳如老狗，告别数据混乱时代！
【8月更文挑战第24天】Apache Kafka作为一款领先的分布式流处理平台，凭借其卓越的高吞吐量与低延迟特性，在大数据处理领域中占据重要地位。然而，在利用Kafka进行数据处理时，如何有效避免重复消费成为众多开发者关注的焦点。本文深入探讨了Kafka中可能出现重复消费的原因，并提出了四种实用的解决方案：利用消息偏移量手动控制消费进度；启用幂等性生产者确保消息不被重复发送；在消费者端实施去重机制；以及借助Kafka的事务支持实现精确的一次性处理。通过这些方法，开发者可根据不同的应用场景灵活选择最适合的策略，从而保障数据处理的准确性和一致性。
土木林森
76 9
你都不懂
|
10天前
|
消息中间件 存储 关系型数据库
Kafka 的基本概念
Kafka 的基本概念
你都不懂
32 0
土木林森
|
1月前
|
消息中间件 负载均衡 Kafka
【Kafka消费秘籍】深入了解消费者组与独立模式，掌握消息消费的两种超能力！
【8月更文挑战第24天】Apache Kafka是一款高性能的分布式消息系统，支持灵活多样的消费模型以适应不同的应用场景。消息按主题组织，每个主题可划分为多个分区，确保消息顺序性。本文深入探讨了Kafka中的两大核心消费模式：消费者组（Consumer Group）和独立消费者（Standalone Consumer）。消费者组允许多个消费者协同工作，实现负载均衡及故障恢复，是最常用的消费模式。独立消费者模式则适用于需要高度定制化处理逻辑的场景，如消息重放等。通过对比这两种模式的特点和提供的示例代码，开发者可以根据具体需求选择最合适的消费策略，从而更好地利用Kafka构建高效的数据流应用程序。
土木林森
36 3
你都不懂
|
1月前
|
消息中间件 存储 Kafka
kafka基础概念二
【8月更文挑战第11天】
你都不懂
35 6
Z_sorrain
|
1月前
|
消息中间件 SQL 分布式计算
Kafka使用场景
Kafka使用场景
Z_sorrain
27 0
JavaSouth南哥
|
2月前
|
消息中间件 存储 Java
全网把Kafka概念讲的最透彻的文章，别无二家
消息队列老大哥Kafka在官网的介绍是这么说的，真是霸气：全球财富前100强公司有超过80%信任并使用Kafka。Kafka目前在GitHub目前也已经有star数27.6k、fork数13.6k。大家好，我是南哥。一个对Java程序员进阶成长颇有研究的人，今天我们开启新的一篇Java进阶指南，本期的对象是Kafka。Kafka历史Star趋势图本文收录在我开源的《Java学习面试指南》中，一份涵盖Java程序员所需掌握核心知识、面试重点的Java指南。希望收到大家的 ⭐ Star ⭐支持。
JavaSouth南哥
101 1
全网把Kafka概念讲的最透彻的文章，别无二家
码农阿豪
|
2月前
|
消息中间件 监控 Java
Kafka 新的消费组默认的偏移量设置和消费行为
Kafka 新的消费组默认的偏移量设置和消费行为
码农阿豪
139 1
不吃核桃
|
2月前
|
消息中间件 Kafka 数据库
面试题Kafka问题之查看偏移量为23的消息如何解决
面试题Kafka问题之查看偏移量为23的消息如何解决
不吃核桃
27 0

热门文章

最新文章

  • 1
    解决kafka集群由于默认的__consumer_offsets这个topic的默认的副本数为1而存在的单点故障问题
  • 2
    SQLServer CDC数据通过Kafka connect实时同步至分析型数据库 AnalyticDB For PostgreSQL及OSS
  • 3
    springboot配置kafka生产者和消费者详解
  • 4
    kafka集群内复制功能深入剖析
  • 5
    Kafka实战－实时日志统计流程
  • 6
    深入理解 Kafka 集群搭建与管理
  • 7
    如何为Kafka加上账号密码（一）
  • 8
    深入理解 Kafka 集群管理与最佳实践
  • 9
    17 个方面，综合对比 Kafka、RabbitMQ、RocketMQ、ActiveMQ 四个分布式消息队列
  • 10
    flume-kafka 实例 详细
  • 1
    扒开kafka内部组件，咱瞅一瞅都有啥？
    59
  • 2
    DataWorks操作报错合集之通过dataworks把表数据发送到kafka，表字段全是string类型如何解决
    60
  • 3
    人工智能平台PAI 操作报错合集之pyalink 1.6.1StreamOperator.fromDataframe(out_df, out_schema_str)之后直接连kafka sink会报下面的错误如何解决
    229
  • 4
    [AIGC 大数据基础] 大数据流处理 Kafka
    42
  • 5
    Python面试：消息队列（RabbitMQ、Kafka）基础知识与应用
    143
  • 6
    Kafka 的生产者优秀架构设计
    62
  • 7
    深度探索：使用Apache Kafka构建高效Java消息队列处理系统
    98
  • 8
    【Kafka】Kafka 的日志保留期与数据清理策略
    1558
  • 9
    【Kafka】Kafka 创建Topic后如何将分区放置到不同的 Broker 中？
    248
  • 10
    【Kakfa】Kafka 的Topic中 Partition 数据是怎么存储到磁盘的？
    229

    • 相关课程

    更多
  • 消息队列Kafka入门课程
  • 分布式消息系统 Kafka 快速入门
  • 消息队列 RocketMQ 全类型业务消息学习课程
  • 消息队列 RocketMQ 消息集成
  • 消息队列RocketMQ 5.0 云原生架构升级课程
  • 带您快速了解阿里云消息队列RocketMQ 5.0版

    • 相关电子书

    更多
  • Java Spring Boot开发实战系列课程【第16讲】：Spring Boot 2.0 实战Apache Kafka百万级高并发消息中间件与原理解析
  • MaxCompute技术公开课第四季 之 如何将Kafka数据同步至MaxCompute
  • 消息队列kafka介绍

    • 相关实验场景

    更多
  • 快速入门轻量消息队列（原MNS）
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 基于ROS的消息发布与订阅
  • 3分钟创建Serverless Job 定时获取新闻热搜
    • 下一篇
    使用OSS上传下载文件