kafka环境搭建以及基本原理

这篇具有很好参考价值的文章主要介绍了kafka环境搭建以及基本原理。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

kafka最先是作为日志数据采集,后用于消息传递,kafka能承担tb级别数据存储,确保服务的可用性,允许少量数据的丢失

作为消息中间件就有异步、解耦、削峰三个作用

一、单机搭建

单机ip:192.168.64.133

下载地址:Apache Kafka 选择kafka_2.13-3.4.0.tgz进行下载

关于kafka的版本,前面的2.13是开发kafka的scala语言的版本,后面的3.4.0是kafka应用的版本。

下载Zookeeper,下载地址 Apache ZooKeeper ,kafka有内置的zookeeper,Zookeeper的版本并没有强制要求,这里我们选择比较新的3.6.1版本。

#下载解压
cd /usr/local/kafka
wget https://archive.apache.org/dist/kafka/3.4.0/kafka_2.13-3.4.0.tgz
tar -zxvf kafka_2.13-3.4.0.tgz

 1、启动Kafka之前需要先启动Zookeeper。**这里就用Kafka自带的Zookeeper。启动脚本在bin目录下。

cd kafka_2.13-3.4.0/
nohup bin/zookeeper-server-start.sh config/zookeeper.properties & 

​ 从nohup.out中可以看到zookeeper默认会在2181端口启动。通过jps指令看到一个QuorumPeerMain进程,确定服务启动成功。 

 2、启动Kafka。

nohup bin/kafka-server-start.sh config/server.properties &

启动完成后,使用jps指令,看到一个kafka进程,确定服务启动成功。服务会默认在9092端口启动。

3、简单收发消息

​ Kafka的基础工作机制是消息发送者可以将消息发送到kafka上指定的topic,而消息消费者,可以从指定的topic上消费消息。

  首先,可以使用Kafka提供的客户端脚本创建Topic

#查看帮助命令
bin/kafka-topics.sh --help
#创建Topic
bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092
#查看Topic
bin/kafka-topics.sh --describe --topic test --bootstrap-server localhost:9092

kafka环境搭建以及基本原理,mq,kafka,分布式 然后,启动一个消息发送者端。往一个名为test的Topic发送消息。

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

当命令行出现 > 符号后,随意输入一些字符然后enter。Ctrl+C 退出命令行。这样就完成了往kafka发消息的操作。

  然后启动一个消息消费端,从名为test的Topic上接收消息。

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test

4、其他消费模式

​ 之前我们通过kafka提供的生产者和消费者脚本,启动了一个简单的消息生产者以及消息消费者,实际上,kafka还提供了丰富的消息消费方式。

指定消费进度

​ 通过kafka-console.consumer.sh启动的控制台消费者,会将获取到的内容在命令行中输出。如果想要消费之前发送的消息,可以通过添加--from-begining参数指定。

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic test

如果需要更精确的消费消息,甚至可以指定从哪一条消息开始消费。 

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --partition 0 --offset 4 --topic test

这表示从第0号Partition上的第四个消息开始读起。Partition和Offset是什么呢,可以用以下指令查看。

分组消费

对于每个消费者,可以指定一个消费者组。kafka中的同一条消息,只能被同一个消费者组下的某一个消费者消费。而不属于同一个消费者组的其他消费者,也可以消费到这一条消息。在kafka-console-consumer.sh脚本中,可以通过--consumer-property group.id=testGroup来指定所属的消费者组。例如,可以启动三个消费者组,来验证一下分组消费机制: 

#两个消费者实例属于同一个消费者组
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --consumer-property group.id=testGrroup --topic test
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --consumer-property group.id=testGrroup --topic test
#这个消费者实例属于不同的消费者组
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --consumer-property group.id=testGrroup2 --topic test

查看消费者组的偏移量

​ 接下来,还可以使用kafka-consumer-groups.sh观测消费者组的情况。包括他们的消费进度。

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group testGroup

kafka环境搭建以及基本原理,mq,kafka,分布式

二、集群搭建

集群就为了保证服务的高可用和数据的安全性,这就是我的理解,万一挂了其它节点依旧可以对外提供服务

1、部署zookeeper集群

#下载解压
cd /usr/local/zookeeper
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.6.1/apache-zookeeper-3.6.1-bin.tar.gz
tar -zxvf apache-zookeeper-3.6.1-bin.tar.gz
#配置
cd /usr/local/zookeeper/apache-zookeeper-3.6.1-bin/conf
cp zoo_sample.cfg zoo.cfg
vim zoo.cfg


#zoo.cfg文件中修改如下配置
#Zookeeper的本地数据目录,默认是/tmp/zookeeper。这是Linux的临时目录,随时会被删掉。
dataDir=/usr/local/zookeeper/data
dataLogDir=/usr/local/zookeeper/log
#Zookeeper的服务端口
clientPort=2181
#集群节点配置
server.1=192.168.64.133:2888:3888
server.2=192.168.64.134:2888:3888
server.3=192.168.64.128:2888:3888
#切换到dataDir目录下
vim myid
#myid文件内容:比如当前节点是server.x,那么内容就是x

其中,clientPort 2181是对客户端开放的服务端口。

集群配置部分, server.x这个x就是节点在集群中的myid。后面的2888端口是集群内部数据传输使用的端口。3888是集群内部进行选举使用的端口。

​ 接下来将整个Zookeeper的应用目录分发到另外两台机器上。就可以在三台机器上都启动Zookeeper服务了。

#切换到zookeeper的home目录下,运行如下命令启动
bin/zkServer.sh --config conf start

启动完成后,使用jps指令可以看到一个QuorumPeerMain进程就表示服务启动成功。

​ 三台机器都启动完成后,可以查看下集群状态。

root@gaorufeng20221210:/usr/local/zookeeper/apache-zookeeper-3.6.1-bin# bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/apache-zookeeper-3.6.1-bin/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader
root@gaorufeng20221210:/usr/local/zookeeper/apache-zookeeper-3.6.1-bin#

这其中Mode 为leader就是主节点,follower就是从节点。

2、部署kafka集群

​ kafka服务并不需要进行选举,因此也没有奇数台服务的建议。

​ 部署Kafka的方式跟部署Zookeeper差不多,就是解压、配置、启服务三板斧。

​ 首先将Kafka解压到/usr/local/kafka目录下。

​ 然后进入config目录,修改server.properties。这个配置文件里面的配置项非常多,下面列出几个要重点关注的配置。

#broker 的全局唯一编号,不能重复,只能是数字。
broker.id=0
#监听端口,要么就配置hosts文件
listeners=PLAINTEXT://192.168.64.133:9092
#数据文件地址。同样默认是给的/tmp目录。
log.dirs=/usr/local/kafka/kafka-logs
#默认的每个Topic的分区数
num.partitions=1
#zookeeper的服务地址
zookeeper.connect=192.168.64.133:2181,192.168.64.134:2181,192.168.64.128:2181
#可以选择指定zookeeper上的基础节点。
#zookeeper.connect=192.168.64.133:2181,192.168.64.134:2181,192.168.64.128:2181/kafka

broker.id需要每个服务器上不一样,分发到其他服务器上时,要注意修改一下。

多个Kafka服务注册到同一个zookeeper集群上的节点,会自动组成集群。

配置文件中的注释非常细致,可以关注一下。下面是server.properties文件中比较重要的核心配置

Property Default Description
broker.id 0 broker的“名字”,你可以选择任意你喜欢的数字作为id,只要id是唯每个broker都可以用一个唯一的非负整数id进行标识;这个id可以作为一的即可。
log.dirs /tmp/kafka-logs kafka存放数据的路径。这个路径并不是唯一的,可以是多个,路径之间只需要使用逗号分隔即可;每当创建新partition时,都会选择在包含最少partitions的路径下进行。
listeners PLAINTEXT://127.0.0.1:9092 server接受客户端连接的端口,ip配置kafka本机ip即可
zookeeper.connect localhost:2181 zookeeper连接地址。hostname:port。如果是Zookeeper集群,用逗号连接。
log.retention.hours 168 每个日志文件删除之前保存的时间。
num.partitions 1 创建topic的默认分区数
default.replication.factor 1 自动创建topic的默认副本数量
min.insync.replicas 1 当producer设置acks为-1时,min.insync.replicas指定replicas的最小数目(必须确认每一个repica的写数据都是成功的),如果这个数目没有达到,producer发送消息会产生异常
delete.topic.enable false 是否允许删除主题

​ 接下来就可以启动kafka服务了。启动服务时需要指定配置文件。

bin/kafka-server-start.sh -daemon config/server.properties

-daemon表示后台启动kafka服务,这样就不会占用当前命令窗口。

​ 通过jps指令可以查看Kafka的进程。

三、Kraft集群

​ 在Kafka的config目录下,提供了一个kraft的文件夹,在这里面就是Kraft协议的参考配置文件。在这个文件夹中有三个配置文件,broker.properties,controller.properties,server.properties,分别给出了Kraft中三种不同角色的示例配置。

  • broker.properties: 数据节点
  • controller.properties: Controller控制节点
  • server.properties: 即可以是数据节点,又可以是Controller控制节点。

这里同样列出几个比较关键的配置项,按照自己的环境进行定制即可

#配置当前节点的角色。Controller相当于Zookeeper的功能,负责集群管理。Broker提供具体的消息转发服务。
process.roles=broker,controller
#配置当前节点的id。与普通集群一样,要求集群内每个节点的ID不能重复。
node.id=1
#配置集群的投票节点。其中@前面的是节点的id,后面是节点的地址和端口,这个端口跟客户端访问的端口是不一样的。通常将集群内的所有Controllor节点都配置进去。
controller.quorum.voters=1@worker1:9093,2@worker2:9093,3@worker3:9093
#Broker对客户端暴露的服务地址。基于PLAINTEXT协议。
advertised.listeners=PLAINTEXT://worker1:9092
#Controller服务协议的别名。默认就是CONTROLLER
controller.listener.names=CONTROLLER
#配置监听服务。不同的服务可以绑定不同的接口。这种配置方式在端口前面是省略了一个主机IP的,主机IP默认是使用的java.net.InetAddress.getCanonicalHostName()
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
#数据文件地址。默认配置在/tmp目录下。
log.dirs=/app/kafka/kraft-log
#topic默认的partition分区数。
num.partitions=2

将配置文件分发,并修改每个服务器上的node.id属性和advertised.listeners属性。

​ 由于Kafka的Kraft集群对数据格式有另外的要求,所以在启动Kraft集群前,还需要对日志目录进行格式化。

[oper@worker1 kafka_2.13-3.4.0]$ bin/kafka-storage.sh random-uuid
j8XGPOrcR_yX4F7ospFkTA
[oper@worker1 kafka_2.13-3.4.0]$ bin/kafka-storage.sh format -t j8XGPOrcR_yX4F7ospFkTA -c config/kraft/server.properties 
Formatting /app/kafka/kraft-log with metadata.version 3.4-IV0.

-t 表示集群ID,三个服务器上可以使用同一个集群ID。

​ 接下来就可以指定配置文件,启动Kafka的服务了。 例如,在Worker1上,启动Broker和Controller服务。

[oper@worker1 kafka_2.13-3.4.0]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties 
[oper@worker1 kafka_2.13-3.4.0]$ jps
10993 Jps
10973 Kafka

​ 等三个服务都启动完成后,就可以像普通集群一样去创建Topic,并维护Topic的信息了。文章来源地址https://www.toymoban.com/news/detail-729196.html

到了这里,关于kafka环境搭建以及基本原理的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • golang—kafka架构原理快速入门以及自测环境搭建(docker单节点部署)

    Apache Kafka 是一个分布式的流处理平台。它具有以下特点: 支持消息的发布和订阅,类似于 RabbtMQ、ActiveMQ 等消息队列 支持数据实时处理 能保证消息的可靠性投递 支持消息的持久化存储,并通过多副本分布式的存储方案来保证消息的容错 高吞吐率,单 Broker 可以轻松处理数千

    2024年02月04日
    浏览(52)
  • 【分布式应用】kafka集群、Filebeat+Kafka+ELK搭建

    主要原因是由于在高并发环境下,同步请求来不及处理,请求往往会发生阻塞。比如大量的请求并发访问数据库,导致行锁表锁,最后请求线程会堆积过多,从而触发 too many connection 错误,引发雪崩效应。 我们使用消息队列,通过异步处理请求,从而缓解系统的压力。消息队

    2024年02月16日
    浏览(45)
  • ZooKeeper之分布式环境搭建--仲裁模式与伪分布式环境搭建

    相关知识 为了完成本关任务,你需要掌握:1.ZooKeeper单节点安装方法,2.命令行基本操作。 ZooKeeper之仲裁模式 standlone 模式运行ZooKeeper,便于评估,开发,测试和学习。但是在实际生产中,使用ZooKeeper均以仲裁模式( quorum mode )运行, quorum mode 具有一组ZooKeeper服务器,这一组

    2024年02月05日
    浏览(47)
  • Spark单机伪分布式环境搭建、完全分布式环境搭建、Spark-on-yarn模式搭建

    搭建Spark需要先配置好scala环境。三种Spark环境搭建互不关联,都是从零开始搭建。 如果将文章中的配置文件修改内容复制粘贴的话,所有配置文件添加的内容后面的注释记得删除,可能会报错。保险一点删除最好。 上传安装包解压并重命名 rz上传 如果没有安装rz可以使用命

    2024年02月06日
    浏览(75)
  • docker搭建Elk+Kafka+Filebeat分布式日志收集系统

    目录 一、介绍 二、集群环境 三、ES集群 四、Kibana  五、Logstash 六、Zookeeper 七、Kafka 八、Filebeat 八、Nginx (一)架构图  (二)组件介绍 1.Elasticsearch 是一个基于Lucene的搜索服务器。提供搜集、分析、存储数据三大功能。它提供了一个分布式多用户能力的全文搜索引擎,基于

    2024年02月04日
    浏览(50)
  • Zookeeper实战——分布式锁实现以及原理

    分布式锁是控制分布式系统之间同步访问共享资源的一种方式。分布式锁的实现方式有很多种,比如 Redis 、数据库 、 zookeeper 等。这篇文章主要介绍用 Zookeeper 实现分布式锁。 先说结论: Zookeeper 是基于临时顺序节点以及 Watcher 监听器机制实现分布式锁的 。 (1)ZooKeeper 的每

    2023年04月08日
    浏览(33)
  • ZooKeeper 分布式协调服务: 概述及原理, 安装配置, 基本操作

    作者:禅与计算机程序设计艺术 Apache Zookeeper 是 Apache Hadoop 的子项目之一,是一个开源的分布式协调服务。它负责存储和维护关于网络中各个节点的数据。Zookeeper 提供了以下功能:配置维护、域名服务、同步和共享、软/硬件负载均衡、集群管理、Master 选举等。它的架构使得

    2024年02月08日
    浏览(39)
  • HBase 伪分布式环境搭建 - 头歌

    mkdir /app cd /opt tar -zxvf hbase-2.1.1-bin.tar.gz -C /app cd /app cd hbase-2.1.1 cd conf echo $JAVA_HOME /usr/lib/jvm/jdk1.8.0_111 vim hbase-env.sh 进入文件,找到下图中红色框框内的路径,将#去掉,把=号后面的路径改成/usr/lib/jvm/jdk1.8.0_111。注意:记得先按A,才能改。 按esc键,输入  :wq  ,回车。 vim hb

    2024年04月28日
    浏览(38)
  • Selenium Grid分布式测试环境搭建

    Selenium Grid简介 Selenium Grid实际上是基于Selenium RC的,而所谓的分布式结构就是由一个hub节点和若干个node代理节点组成。Hub用来管理各个代理节点的注册信息和状态信息,并且接受远程客户端代码的请求调用,然后把请求的命令转发给代理节点来执行(官方说明) 简单来说通过

    2024年02月03日
    浏览(36)
  • ZooKeeper+HBase分布式集群环境搭建

    安装版本:hadoop-2.10.1、zookeeper-3.4.12、hbase-2.3.1 一、zookeeper 集群搭建与配置 1. 下载zookeeper安装包 2. 解压移动zookeeper 3. 修改配置文件(创建文件夹) 4. 进入conf/ 5. 修改zoo.cfg文件 6. 进入/usr/local/zookeeper-3.4.12/zkdatas/这个路径下创建一个文件,文件名为myid ,文件内容为1 7. 拷贝到

    2024年02月08日
    浏览(44)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包