草庐IT

kafka 基础概念、命令行操作(查看所有topic、创建topic、删除topic、查看某个Topic的详情、修改分区数、发送消息、消费消息、 查看消费者组 、更新消费者的偏移位置)

但行益事莫问前程 2023-12-31 原文

文章目录

前言

kafka官网


1. 基础概念

Broker

  一台kafka服务器就是一个broker,可容纳多个topic。一个集群由多个broker组成;

Producer

  生产者,即向kafka的broker-list发送消息的客户端;

Consumer

  消费者,即向kafka的broker-list订阅消息的客户端;

Consumer Group

  消费者组是逻辑上的一个订阅者,由多个consumer组成。消费者组内每个消费者负责消费不同分区的数据,一个分区只能由一个组内消费者消费;所有的消费者都属于某个消费者组,消费者组之间互不影响;

Topic

  可理解为一个队列,生产者和消费者均面向一个topic;

Partition

  一个topic可以分为多个partition,每个partition是一个有序的队列;这使得一个topic可分布到多个broker(即服务器)上,

Replica

  副本,kafka提供了副本机制,一个topic的每个分区都有若干个副本,即一个leader和若干个follower,从而保证集群单点故障时节点上的partition数据不丢失
leader:每个分区多个副本的生产者发送数据的对象,以及消费者消费数据的对象都是leader
follower:实时从leader中同步数据,保持和leader数据的同步。leader发生故障时,某个follower会成为新的leader


2. 命令行操作

2.1 查看所有topic

./kafka-topics.sh --bootstrap-server 192.168.0.23:29092,192.168.0.114:29092,192.168.0.110:29092 --list

2.2 创建topic

./kafka-topics.sh --bootstrap-server 192.168.0.23:29092,192.168.0.110:29092 --create --replication-factor 3 --partitions 3 --topic 
charges

--topic 定义topic名
--replication-factor 定义副本数
--partitions 定义分区数

2.3 删除topic

./kafka-topics.sh --bootstrap-server 192.168.0.23:29092,192.168.0.114:29092,192.168.0.110:29092 --delete --topic charges

2.4 查看某个Topic的详情

./kafka-topics.sh --bootstrap-server 192.168.0.23:29092,192.168.0.114:29092,192.168.0.110:29092 --describe --topic charges

2.5 修改分区数

./kafka-topics.sh --bootstrap-server 192.168.0.23:29092,192.168.0.114:29092,192.168.0.110:29092 --alter --partitions 6 --topic charges


2.6 发送消息

./kafka-console-producer.sh --broker-list 192.168.0.23:29092,192.168.0.114:29092,192.168.0.110:29092 --topic charges


2.7 消费消息

./kafka-console-consumer.sh --bootstrap-server 192.168.0.23:29092,192.168.0.114:29092,192.168.0.110:29092 --from-beginning  --topic charges --group htsc

--from-beginning:若消费者没有offset,则从 the earliest message present in the log 而非默认的 the latest message
--group : 指定消费者所属的消费者组

注:默认情况下,若zookeeper或__consumer_offsets中有offset记录,则Kafka消费者从最后一次提交的偏移量位置(offset)开始消费消息。若之前没有提交过偏移量,则消费消息开始位置取决于Topic的配置属性auto.offset.reset的设置,默认为最新(latest 仅消费新消息


2.8 查看消费者组

./kafka-consumer-groups.sh --bootstrap-server 192.168.0.23:29092,192.168.0.114:29092,192.168.0.110:29092 --list

./kafka-consumer-groups.sh --bootstrap-server 192.168.0.23:29092,192.168.0.114:29092,192.168.0.110:29092 --describe --group htsc


注: 通过查看消费者组,可查看到topic所有分区当前的偏移量以及LOG-END-OFFSET(最后的偏移量),即消费消息的现状

2.9 更新消费者的偏移位置

  将topic charges对应的消费组htsc 的分区编号0的偏移量设置为最早

./kafka-consumer-groups.sh --bootstrap-server 192.168.0.23:29092,192.168.0.114:29092,192.168.0.110:29092 --topic charges:0 --group htsc --reset-offsets --to-earliest --execute

--reset-offsets Reset offsets of consumer group

--to-earliest Reset offsets to earliest offset
--to-latest Reset offsets to latest offset.
--to-offset <Long: offset> Reset offsets to a specific offset


有关kafka 基础概念、命令行操作(查看所有topic、创建topic、删除topic、查看某个Topic的详情、修改分区数、发送消息、消费消息、 查看消费者组 、更新消费者的偏移位置)的更多相关文章

  1. ruby - 如何以所有可能的方式将字符串拆分为长度最多为 3 的连续子字符串? - 2

    我试图获取一个长度在1到10之间的字符串,并输出将字符串分解为大小为1、2或3的连续子字符串的所有可能方式。例如:输入:123456将整数分割成单个字符,然后继续查找组合。该代码将返回以下所有数组。[1,2,3,4,5,6][12,3,4,5,6][1,23,4,5,6][1,2,34,5,6][1,2,3,45,6][1,2,3,4,56][12,34,5,6][12,3,45,6][12,3,4,56][1,23,45,6][1,2,34,56][1,23,4,56][12,34,56][123,4,5,6][1,234,5,6][1,2,345,6][1,2,3,456][123

  2. ruby - 在 Ruby 中编写命令行实用程序 - 2

    我想用ruby​​编写一个小的命令行实用程序并将其作为gem分发。我知道安装后,Guard、Sass和Thor等某些gem可以从命令行自行运行。为了让gem像二进制文件一样可用,我需要在我的gemspec中指定什么。 最佳答案 Gem::Specification.newdo|s|...s.executable='name_of_executable'...endhttp://docs.rubygems.org/read/chapter/20 关于ruby-在Ruby中编写命令行实用程序

  3. ruby-on-rails - 跳过状态机方法的所有验证 - 2

    当我的预订模型通过rake任务在状态机上转换时,我试图找出如何跳过对ActiveRecord对象的特定实例的验证。我想在reservation.close时跳过所有验证!叫做。希望调用reservation.close!(:validate=>false)之类的东西。仅供引用,我们正在使用https://github.com/pluginaweek/state_machine用于状态机。这是我的预订模型的示例。classReservation["requested","negotiating","approved"])}state_machine:initial=>'requested

  4. ruby - Nokogiri 剥离所有属性 - 2

    我有这个html标记:我想得到这个:我如何使用Nokogiri做到这一点? 最佳答案 require'nokogiri'doc=Nokogiri::HTML('')您可以通过xpath删除所有属性:doc.xpath('//@*').remove或者,如果您需要做一些更复杂的事情,有时使用以下方法遍历所有元素会更容易:doc.traversedo|node|node.keys.eachdo|attribute|node.deleteattributeendend 关于ruby-Nokog

  5. ruby-on-rails - 如何在 Rails View 上显示错误消息? - 2

    我是rails的新手,想在form字段上应用验证。myviewsnew.html.erb.....模拟.rbclassSimulation{:in=>1..25,:message=>'Therowmustbebetween1and25'}end模拟Controller.rbclassSimulationsController我想检查模型类中row字段的整数范围,如果不在范围内则返回错误信息。我可以检查上面代码的范围,但无法返回错误消息提前致谢 最佳答案 关键是您使用的是模型表单,一种显示ActiveRecord模型实例属性的表单。c

  6. ruby - 获取模块中定义的所有常量的值 - 2

    我想获取模块中定义的所有常量的值:moduleLettersA='apple'.freezeB='boy'.freezeendconstants给了我常量的名字:Letters.constants(false)#=>[:A,:B]如何获取它们的值的数组,即["apple","boy"]? 最佳答案 为了做到这一点,请使用mapLetters.constants(false).map&Letters.method(:const_get)这将返回["a","b"]第二种方式:Letters.constants(false).map{|c

  7. ruby - 使用 Ruby 通过 Outlook 发送消息的最简单方法是什么? - 2

    我的工作要求我为某些测试自动生成电子邮件。我一直在四处寻找,但未能找到可以快速实现的合理解决方案。它需要在outlook而不是其他邮件服务器中,因为我们有一些奇怪的身份验证规则,我们需要保存草稿而不是仅仅发送邮件的选项。显然win32ole可以做到这一点,但我找不到任何相当简单的例子。 最佳答案 假设存储了Outlook凭据并且您设置为自动登录到Outlook,WIN32OLE可以很好地完成此操作:require'win32ole'outlook=WIN32OLE.new('Outlook.Application')message=

  8. Ruby - 如何将消息长度表示为 2 个二进制字节 - 2

    我正在使用Ruby,我正在与一个网络端点通信,该端点在发送消息本身之前需要格式化“header”。header中的第一个字段必须是消息长度,它被定义为网络字节顺序中的2二进制字节消息长度。比如我的消息长度是1024。如何将1024表示为二进制双字节? 最佳答案 Ruby(以及Perl和Python等)中字节整理的标准工具是pack和unpack。ruby的packisinArray.您的长度应该是两个字节长,并且按网络字节顺序排列,这听起来像是n格式说明符的工作:n|Integer|16-bitunsigned,network(bi

  9. ruby-on-rails - 在 Flash 警报 Rails 3 中显示错误消息 - 2

    如果我在模型中设置验证消息validates:name,:presence=>{:message=>'Thenamecantbeblank.'}我如何让该消息显示在闪光警报中,这是我迄今为止尝试过的方法defcreate@message=Message.new(params[:message])if@message.valid?ContactMailer.send_mail(@message).deliverredirect_to(root_path,:notice=>"Thanksforyourmessage,Iwillbeintouchsoon")elseflash[:error]

  10. ruby - 如何使用 Selenium Webdriver 根据 div 的内容执行操作? - 2

    我有一个使用SeleniumWebdriver和Nokogiri的Ruby应用程序。我想选择一个类,然后对于那个类对应的每个div,我想根据div的内容执行一个Action。例如,我正在解析以下页面:https://www.google.com/webhp?sourceid=chrome-instant&ion=1&espv=2&ie=UTF-8#q=puppies这是一个搜索结果页面,我正在寻找描述中包含“Adoption”一词的第一个结果。因此机器人应该寻找带有className:"result"的div,对于每个检查它的.descriptiondiv是否包含单词“adoption

随机推荐