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

简介: 【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廖志伟

🍊社区:幕后大佬

相关文章
|
4月前
|
消息中间件 负载均衡 大数据
Kafka - 分区中各种偏移量的说明
Kafka - 分区中各种偏移量的说明
84 0
Kafka - 分区中各种偏移量的说明
|
1月前
|
消息中间件 Kafka
Kafka - 深入了解Kafka基础架构:Kafka的基本概念
Kafka - 深入了解Kafka基础架构:Kafka的基本概念
16 0
|
1月前
|
消息中间件 存储 Kafka
Kafka - 消息队列的两种模式
Kafka - 消息队列的两种模式
43 0
|
2月前
|
消息中间件 存储 Java
Kafka基本概念与应用场景
Apache Kafka是一种分布式的、基于发布/订阅的消息系统,由Scala语言编写而成。它具备快速、可扩展、可持久化的特点。Kafka最初由LinkedIn开发,并于2011年初开源, 2012年10月从Apache孵化器毕业,成为Apache基金会的顶级项目。
|
5月前
|
消息中间件 传感器 Kafka
(三)kafka从入门到精通之使用场景
Kafka 是一种流处理平台,主要用于处理大量数据流,如实时事件、日志文件和传感器数据等。Kafka的目的是实现高吞吐量、低延迟和高可用性的数据处理。Kafka提供了一个高度可扩展的架构,可以轻松地添加和删除节点,并且能够处理数百亿条消息/分区。Kafka的消息可以容错,即使某个节点失败,消息也会在集群中的其他节点上得到处理。总的来说,Kafka 是一个非常强大的数据处理平台,可以用于实时数据处理、日志文件处理、传感器数据处理和流处理等场景。
42 0
|
6月前
|
消息中间件 存储 Kafka
Kafka架构及基本概念
刚开始了解Kafka时对其中多个名词表示懵逼,broker是啥?咋还有分区?有没有跟和我一样有很多???本文就我对Kafka的理解梳理各个角色以及功能,欢迎大家一起来沟通交流
79 0
|
6月前
|
消息中间件 Kafka
为什么kafka 需要 subscribe 的 group.id?我们是否需要使用 commitSync 手动提交偏移量?
Kafka 使用消费者组的概念来实现主题的并行消费 - 每条消息都将在每个消费者组中传递一次,无论该组中实际有多少个消费者。所以 group 参数是强制性的,如果没有组,Kafka 将不知道如何对待订阅同一主题的其他消费者。
79 2
|
9月前
|
消息中间件 缓存 容灾
Apache Kafka-通过设置Consumer Group实现广播模式
Apache Kafka-通过设置Consumer Group实现广播模式
1375 0
|
9月前
|
消息中间件 分布式计算 Java
「事件驱动架构」Kafka中的模式注册表和模式演化
「事件驱动架构」Kafka中的模式注册表和模式演化
|
消息中间件 Kafka
Kafka单机模式和集群模式环境搭建
Kafka单机模式和集群模式环境搭建
118 0

相关产品

  • 云迁移中心