最新要闻
- 索尼被曝测试新版PS5:性能不变、增设可拆卸光驱
- 全球热消息:因担心投资者被吸引到他国:欧盟再次推迟将锂列为有害物质
- 世界今亮点!苹果担心的事儿发生了 男子用AirTag追踪妻子被逮捕
- 《巫师3》次时代版体积惊人:PS5升级包超50GB
- 【世界播资讯】国家邮政局官方回应“快递不快”:加快恢复 邮政顺丰京东开启夜派
- 锵锵三人行停播原因是什么?锵锵三人行女嘉宾名单大全
- 沙海吴邪的计划是什么?沙海吴邪的计划成功了吗?
- 什么边野草花什么口夕阳斜?什么边野草花什么口夕阳斜是什么诗?
- 九月青豆角产自什么村?九月青豆角怎么冻好吃?
- 天国的阶梯结局是什么?天国的阶梯演员表
- 天天看点:卡梅隆感染新冠 缺席洛杉矶首映式
- 想“白嫖”20万的私人飞机 推特被告了
- 【报资讯】短了5厘米照样强大 迷你SSD硬盘雄起 速度冲向5GB/s
- 全球聚焦:小米13、iPhone 14 Pro全角度对比:小米正面碾压式完胜 背面有争议
- 环球快消息!世界杯半决赛现场将播放两首中文歌:你肯定都听过
- 环球播报:硬刚小米13!moto X40核心配置官方全面揭晓
手机
iphone11大小尺寸是多少?苹果iPhone11和iPhone13的区别是什么?
警方通报辅警执法直播中被撞飞:犯罪嫌疑人已投案
- iphone11大小尺寸是多少?苹果iPhone11和iPhone13的区别是什么?
- 警方通报辅警执法直播中被撞飞:犯罪嫌疑人已投案
- 男子被关545天申国赔:获赔18万多 驳回精神抚慰金
- 3天内26名本土感染者,辽宁确诊人数已超安徽
- 广西柳州一男子因纠纷杀害三人后自首
- 洱海坠机4名机组人员被批准为烈士 数千干部群众悼念
家电
基于zookeeper的kafka中间件
一、Zookeeper 概述
1.1Zookeeper 定义
Zookeeper是一个开源的分布式的,为分布式框架提供协调服务的Apache项目。
1.2Zookeeper 工作机制
Zookeeper从设计模式角度来理解:是一个基于观察者模式设计的分布式服务管理框架,它负责存储和管理大家都关心的数据,然后接受观察者的注册,一旦这些数据的状态发生变化,Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。也就是说 Zookeeper = 文件系统 + 通知机制。
1.3Zookeeper 特点
(1)Zookeeper:一个领导者(Leader),多个跟随者(Follower)组成的集群。(2)Zookeepe集群中只要有半数以上节点存活,Zookeeper集群就能正常服务。所以Zookeeper适合安装奇数台服务器。(3)全局数据一致:每个Server保存一份相同的数据副本,Client无论连接到哪个Server,数据都是一致的。(4)更新请求顺序执行,来自同一个Client的更新请求按其发送顺序依次执行,即先进先出。(5)数据更新原子性,一次数据更新要么成功,要么失败。(6)实时性,在一定时间范围内,Client能读到最新数据。
(资料图片仅供参考)
1.4Zookeeper 数据结构
ZooKeeper数据模型的结构与Linux文件系统很类似,整体上可以看作是一棵树,每个节点称做一个ZNode。每一个ZNode默认能够存储1MB的数据,每个ZNode都可以通过其路径唯一标识。
1.5Zookeeper 应用场景
提供的服务包括:统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。●统一命名服务在分布式环境下,经常需要对应用/服务进行统一命名,便于识别。例如:IP不容易记住,而域名容易记住。●统一配置管理(1)分布式环境下,配置文件同步非常常见。一般要求一个集群中,所有节点的配置信息是一致的,比如Kafka集群。对配置文件修改后,希望能够快速同步到各个节点上。(2)配置管理可交由ZooKeeper实现。可将配置信息写入ZooKeeper上的一个Znode。各个客户端服务器监听这个Znode。一旦 Znode中的数据被修改,ZooKeeper将通知各个客户端服务器。●统一集群管理(1)分布式环境中,实时掌握每个节点的状态是必要的。可根据节点实时状态做出一些调整。(2)ZooKeeper可以实现实时监控节点状态变化。可将节点信息写入ZooKeeper上的一个ZNode。监听这个ZNode可获取它的实时状态变化。●服务器动态上下线客户端能实时洞察到服务器上下线的变化。●软负载均衡在Zookeeper中记录每台服务器的访问数,让访问数最少的服务器去处理最新的客户端请求。
1.6Zookeeper 选举机制
(1)第一次启动选举机制
(1)服务器1启动,发起一次选举。服务器1投自己一票。此时服务器1票数一票,不够半数以上(3票),选举无法完成,服务器1状态保持为LOOKING;(2)服务器2启动,再发起一次选举。服务器1和2分别投自己一票并交换选票信息:此时服务器1发现服务器2的myid比自己目前投票推举的(服务器1)大,更改选票为推举服务器2。此时服务器1票数0票,服务器2票数2票,没有半数以上结果,选举无法完成,服务器1,2状态保持LOOKING(3)服务器3启动,发起一次选举。此时服务器1和2都会更改选票为服务器3。此次投票结果:服务器1为0票,服务器2为0票,服务器3为3票。此时服务器3的票数已经超过半数,服务器3当选Leader。服务器1,2更改状态为FOLLOWING,服务器3更改状态为LEADING;(4)服务器4启动,发起一次选举。此时服务器1,2,3已经不是LOOKING状态,不会更改选票信息。交换选票信息结果:服务器3为3票,服务器4为1票。此时服务器4服从多数,更改选票信息为服务器3,并更改状态为FOLLOWING;(5)服务器5启动,同4一样当小弟。
(2)非第一次启动选举机制
1、当ZooKeeper 集群中的一台服务器出现以下两种情况之一时,就会开始进入Leader选举:(1)服务器初始化启动。(2)服务器运行期间无法和Leader保持连接。2、而当一台机器进入Leader选举流程时,当前集群也可能会处于以下两种状态:(1)集群中本来就已经存在一个Leader。对于已经存在Leader的情况,机器试图去选举Leader时,会被告知当前服务器的Leader信息,对于该机器来说,仅仅需要和 Leader机器建立连接,并进行状态同步即可。(2)集群中确实不存在Leader。假设ZooKeeper由5台服务器组成,SID分别为1、2、3、4、5,ZXID分别为8、8、8、7、7,并且此时SID为3的服务器是Leader。某一时刻,3和5服务器出现故障,因此开始进行Leader选举。
选举Leader规则:1.EPOCH大的直接胜出2.EPOCH相同,事务id大的胜出3.事务id相同,服务器id大的胜出注释:SID:服务器ID。用来唯一标识一台ZooKeeper集群中的机器,每台机器不能重复,和myid一致。ZXID:事务ID。ZXID是一个事务ID,用来标识一次服务器状态的变更。在某一时刻,集群中的每台机器的ZXID值不一定完全一致,这和ZooKeeper服务器对于客户端“更新请求”的处理逻辑速度有关。Epoch:每个Leader任期的代号。没有Leader时同一轮投票过程中的逻辑时钟值是相同的。每投完一次票这个数据就会增加
二、部署 Zookeeper 集群
2.1部署Zookeeper集群的操作步骤
准备 3 台服务器做 Zookeeper 集群192.168.10.10192.168.10.20192.168.10.30
2.1.1 安装前准备
@关闭防火墙
systemctl stop firewalldsystemctl disable firewalldsetenforce 0
@安装 JDK
yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-develjava -version
@下载安装包官方下载地址:https://archive.apache.org/dist/zookeeper/
cd /optwget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.7/apache-zookeeper-3.5.7-bin.tar.gz
2.1.2安装 Zookeeper
cd /opttar -zxvf apache-zookeeper-3.5.7-bin.tar.gzmv apache-zookeeper-3.5.7-bin /usr/local/zookeeper-3.5.7
2.1.3修改配置文件
cd /usr/local/zookeeper-3.5.7/conf/cp zoo_sample.cfg zoo.cfg vim zoo.cfgtickTime=2000 #通信心跳时间,Zookeeper服务器与客户端心跳时间,单位毫秒initLimit=10 #Leader和Follower初始连接时能容忍的最多心跳数(tickTime的数量),这里表示为10*2ssyncLimit=5 #Leader和Follower之间同步通信的超时时间,这里表示如果超过5*2s,Leader认为Follwer死掉,并从服务器列表中删除FollwerdataDir=/usr/local/zookeeper-3.5.7/data ●修改,指定保存Zookeeper中的数据的目录,目录需要单独创建dataLogDir=/usr/local/zookeeper-3.5.7/logs ●添加,指定存放日志的目录,目录需要单独创建clientPort=2181 #客户端连接端口#添加集群信息server.1=192.168.10.10:3188:3288server.2=192.168.10.20:3188:3288server.3=192.168.10.30:3188:3288
server.A=B:C:D
- A是一个数字,表示这个是第几号服务器。集群模式下需要在zoo.cfg中dataDir指定的目录下创建一个文件myid,这个文件里面有一个数据就是A的值,Zookeeper启动时读取此文件,拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。
- B是这个服务器的地址。
- C是这个服务器Follower与集群中的Leader服务器交换信息的端口。
- D是万一集群中的Leader服务器挂了,需要一个端口来重新进行选举,选出一个新的Leader,而这个端口就是用来执行选举时服务器相互通信的端口。
2.1.4拷贝配置好的 Zookeeper 配置文件到其他机器上
scp /usr/local/zookeeper-3.5.7/conf/zoo.cfg 192.168.10.20:/usr/local/zookeeper-3.5.7/conf/scp /usr/local/zookeeper-3.5.7/conf/zoo.cfg 192.168.20.30:/usr/local/zookeeper-3.5.7/conf/
2.1.5在每个节点上创建数据目录和日志目录
mkdir /usr/local/zookeeper-3.5.7/datamkdir /usr/local/zookeeper-3.5.7/logs
2.1.6在每个节点的dataDir指定的目录下创建一个 myid 的文件
echo 1 > /usr/local/zookeeper-3.5.7/data/myidecho 2 > /usr/local/zookeeper-3.5.7/data/myidecho 3 > /usr/local/zookeeper-3.5.7/data/myid
2.1.7配置 Zookeeper 启动脚本
vim /etc/init.d/zookeeper#!/bin/bash#chkconfig:2345 20 90#description:Zookeeper Service Control ScriptZK_HOME="/usr/local/zookeeper-3.5.7"case $1 instart)echo "---------- zookeeper 启动 ------------"$ZK_HOME/bin/zkServer.sh start;;stop)echo "---------- zookeeper 停止 ------------"$ZK_HOME/bin/zkServer.sh stop;;restart)echo "---------- zookeeper 重启 ------------"$ZK_HOME/bin/zkServer.sh restart;;status)echo "---------- zookeeper 状态 ------------"$ZK_HOME/bin/zkServer.sh status;;*)echo "Usage: $0 {start|stop|restart|status}"esac
2.1.8设置开机自启
chmod +x /etc/init.d/zookeeperchkconfig --add zookeeper
2.1.9分别启动 Zookeeper
service zookeeper start
2.1.10查看当前状态
service zookeeper status
2.2实例操作:部署Zookeeper集群(三台机器)
2.2.1安装前准备
2.2.2安装 Zookeeper
2.2.3修改配置文件
2.2.4拷贝配置好的 Zookeeper 配置文件到其他机器上
2.2.5在每个节点上创建数据目录和日志目录
2.2.6在每个节点的dataDir指定的目录下创建一个 myid 的文件
2.2.7配置 Zookeeper 启动脚本
2.2.8设置开机自启并开启服务
2.2.9查看当前状态
一个leader,两个follower
三、Kafka 概述
3.1为什么需要消息队列(MQ)
主要原因是由于在高并发环境下,同步请求来不及处理,请求往往会发生阻塞。比如大量的请求并发访问数据库,导致行锁表锁,最后请求线程会堆积过多,从而触发 too many connection 错误,引发雪崩效应。我们使用消息队列,通过异步处理请求,从而缓解系统的压力。消息队列常应用于异步处理,流量削峰,应用解耦,消息通讯等场景。当前比较常见的 MQ 中间件有 ActiveMQ、RabbitMQ、RocketMQ、Kafka 等。
3.2使用消息队列的好处
(1)解耦:允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。(2)可恢复性:系统的一部分组件失效时,不会影响到整个系统。消息队列降低了进程间的耦合度,所以即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。(3)缓冲:有助于控制和优化数据流经过系统的速度,解决生产消息和消费消息的处理速度不一致的情况。(4)灵活性 & 峰值处理能力:在访问量剧增的情况下,应用仍然需要继续发挥作用,但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力,而不会因为突发的超负荷的请求而完全崩溃。(5)异步通信:很多时候,用户不想也不需要立即处理消息。消息队列提供了异步处理机制,允许用户把一个消息放入队列,但并不立即处理它。想向队列中放入多少消息就放多少,然后在需要的时候再去处理它们。
3.3消息队列的两种模式
(1)点对点模式(一对一,消费者主动拉取数据,消息收到后消息清除)消息生产者生产消息发送到消息队列中,然后消息消费者从消息队列中取出并且消费消息。消息被消费以后,消息队列中不再有存储,所以消息消费者不可能消费到已经被消费的消息。消息队列支持存在多个消费者,但是对一个消息而言,只会有一个消费者可以消费。(2)发布/订阅模式(一对多,又叫观察者模式,消费者消费数据之后不会清除消息)消息生产者(发布)将消息发布到 topic 中,同时有多个消息消费者(订阅)消费该消息。和点对点方式不同,发布到 topic 的消息会被所有订阅者消费。发布/订阅模式是定义对象间一种一对多的依赖关系,使得每当一个对象(目标对象)的状态发生改变,则所有依赖于它的对象(观察者对象)都会得到通知并自动更新。
3.4Kafka 定义
Kafka 是一个分布式的基于发布/订阅模式的消息队列(MQ,Message Queue),主要应用于大数据实时处理领域。
3.5Kafka 简介
Kafka 是最初由 Linkedin 公司开发,是一个分布式、支持分区的(partition)、多副本的(replica),基于 Zookeeper 协调的分布式消息中间件系统,它的最大的特性就是可以实时的处理大量数据以满足各种需求场景,比如基于 hadoop 的批处理系统、低延迟的实时系统、Spark/Flink 流式处理引擎,nginx 访问日志,消息服务等等,用 scala 语言编写。Linkedin 于 2010 年贡献给了 Apache 基金会并成为顶级开源项目。
3.6Kafka 的特性
(1)高吞吐量、低延迟Kafka 每秒可以处理几十万条消息,它的延迟最低只有几毫秒。每个 topic 可以分多个 Partition,Consumer Group 对 Partition 进行消费操作,提高负载均衡能力和消费能力。(2)可扩展性kafka 集群支持热扩展(3)持久性、可靠性消息被持久化到本地磁盘,并且支持数据备份防止数据丢失(4)容错性允许集群中节点失败(多副本情况下,若副本数量为 n,则允许 n-1 个节点失败)(5)高并发支持数千个客户端同时读写
3.7Kafka 系统架构
(1)Broker:一台 kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker 可以容纳多个 topic。(2)Topic:可以理解为一个队列,生产者和消费者面向的都是一个 topic。类似于数据库的表名或者 ES 的 index,物理上不同 topic 的消息分开存储(3)Partition:为了实现扩展性,一个非常大的 topic 可以分布到多个 broker(即服务器)上,一个 topic 可以分割为一个或多个 partition,每个 partition 是一个有序的队列。Kafka 只保证 partition 内的记录是有序的,而不保证 topic 中不同 partition 的顺序。每个 topic 至少有一个 partition,当生产者产生数据的时候,会根据分配策略选择分区,然后将消息追加到指定的分区的队列末尾。
Partation 数据路由规则
- 指定了 patition,则直接使用;
- 未指定 patition 但指定 key(相当于消息中某个属性),通过对 key 的 value 进行 hash 取模,选出一个 patition;
- patition 和 key 都未指定,使用轮询选出一个 patition。
补充:每条消息都会有一个自增的编号,用于标识消息的偏移量,标识顺序从 0 开始。每个 partition 中的数据使用多个 segment 文件存储。如果 topic 有多个 partition,消费数据时就不能保证数据的顺序。严格保证消息的消费顺序的场景下(例如商品秒杀、 抢红包),需要将 partition 数目设为 1。
- broker 存储 topic 的数据。如果某 topic 有 N 个 partition,集群有 N 个 broker,那么每个 broker 存储该 topic 的一个 partition。
- 如果某 topic 有 N 个 partition,集群有 (N+M) 个 broker,那么其中有 N 个 broker 存储 topic 的一个 partition, 剩下的 M 个 broker 不存储该 topic 的 partition 数据。
- 如果某 topic 有 N 个 partition,集群中 broker 数目少于 N 个,那么一个 broker 存储该 topic 的一个或多个 partition。在实际生产环境中,尽量避免这种情况的发生,这种情况容易导致 Kafka 集群数据不均衡。
分区的原因
- 方便在集群中扩展,每个Partition可以通过调整以适应它所在的机器,而一个topic又可以有多个Partition组成,因此整个集群就可以适应任意大小的数据了;
- 可以提高并发,因为可以以Partition为单位读写了。
(4)Replica:副本,为保证集群中的某个节点发生故障时,该节点上的 partition 数据不丢失,且 kafka 仍然能够继续工作,kafka 提供了副本机制,一个 topic 的每个分区都有若干个副本,一个 leader 和若干个 follower。(5)Leader:每个 partition 有多个副本,其中有且仅有一个作为 Leader,Leader 是当前负责数据的读写的 partition。(6)Follower:Follower 跟随 Leader,所有写请求都通过 Leader 路由,数据变更会广播给所有 Follower,Follower 与 Leader 保持数据同步。Follower 只负责备份,不负责数据的读写。如果 Leader 故障,则从 Follower 中选举出一个新的 Leader。当 Follower 挂掉、卡住或者同步太慢,Leader 会把这个 Follower 从 ISR(Leader 维护的一个和 Leader 保持同步的 Follower 集合) 列表中删除,重新创建一个 Follower。(7)Producer:生产者即数据的发布者,该角色将消息发布到 Kafka 的 topic 中。broker 接收到生产者发送的消息后,broker 将该消息追加到当前用于追加数据的 segment 文件中。生产者发送的消息,存储到一个 partition 中,生产者也可以指定数据存储的 partition。(8)Consumer:消费者可以从 broker 中读取数据。消费者可以消费多个 topic 中的数据。(9)Consumer Group(CG)
- 消费者组,由多个 consumer 组成。
- 所有的消费者都属于某个消费者组,即消费者组是逻辑上的一个订阅者。可为每个消费者指定组名,若不指定组名则属于默认的组。
- 将多个消费者集中到一起去处理某一个 Topic 的数据,可以更快的提高数据的消费能力。
- 消费者组内每个消费者负责消费不同分区的数据,一个分区只能由一个组内消费者消费,防止数据被重复读取。
- 消费者组之间互不影响。
(10)offset 偏移量
- 可以唯一的标识一条消息。
- 偏移量决定读取数据的位置,不会有线程安全的问题,消费者通过偏移量来决定下次读取的消息(即消费位置)。
- 消息被消费之后,并不被马上删除,这样多个业务就可以重复使用 Kafka 的消息。
- 某一个业务也可以通过修改偏移量达到重新读取消息的目的,偏移量由用户控制。
- 消息最终还是会被删除的,默认生命周期为 1 周(7*24小时)。
(11)Zookeeper
- Kafka 通过 Zookeeper 来存储集群的 meta 信息。
- 由于 consumer 在消费过程中可能会出现断电宕机等故障,consumer 恢复后,需要从故障前的位置的继续消费,所以 consumer 需要实时记录自己消费到了哪个 offset,以便故障恢复后继续消费。
- Kafka 0.9 版本之前,consumer 默认将 offset 保存在 Zookeeper 中;从 0.9 版本开始,consumer 默认将 offset 保存在 Kafka 一个内置的 topic 中,该 topic 为 __consumer_offsets。
四、部署zookeeper + kafka 集群
4.1部署zookeeper + kafka 集群
4.1.1下载安装包
官方下载地址:http://kafka.apache.org/downloads.html
cd /optwget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz
4.1.2安装 Kafka
cd /opt/tar zxvf kafka_2.13-2.7.1.tgzmv kafka_2.13-2.7.1 /usr/local/kafka
4.1.3修改配置文件
cd /usr/local/kafka/config/cp server.properties{,.bak} vim server.propertiesbroker.id=0 #21行,broker的全局唯一编号,每个broker不能重复,因此要在其他机器上配置 broker.id=1、broker.id=2listeners=PLAINTEXT://192.168.10.10:9092 #31行,指定监听的IP和端口,如果修改每个broker的IP需区分开来,也可保持默认配置不用修改num.network.threads=3 #42行,broker 处理网络请求的线程数量,一般情况下不需要去修改num.io.threads=8 #45行,用来处理磁盘IO的线程数量,数值应该大于硬盘数socket.send.buffer.bytes=102400 #48行,发送套接字的缓冲区大小socket.receive.buffer.bytes=102400 #51行,接收套接字的缓冲区大小socket.request.max.bytes=104857600 #54行,请求套接字的缓冲区大小log.dirs=/usr/local/kafka/logs #60行,kafka运行日志存放的路径,也是数据存放的路径num.partitions=1 #65行,topic在当前broker上的默认分区个数,会被topic创建时的指定参数覆盖num.recovery.threads.per.data.dir=1 #69行,用来恢复和清理data下数据的线程数量log.retention.hours=168 #103行,segment文件(数据文件)保留的最长时间,单位为小时,默认为7天,超时将被删除log.segment.bytes=1073741824 #110行,一个segment文件最大的大小,默认为 1G,超出将新建一个新的segment文件zookeeper.connect=192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181 #123行,配置连接Zookeeper集群地址
4.1.4修改环境变量
vim /etc/profileexport KAFKA_HOME=/usr/local/kafkaexport PATH=$PATH:$KAFKA_HOME/bin source /etc/profile
4.1.5配置 Zookeeper 启动脚本
vim /etc/init.d/kafka#!/bin/bash#chkconfig:2345 22 88#description:Kafka Service Control ScriptKAFKA_HOME="/usr/local/kafka"case $1 instart)echo "---------- Kafka 启动 ------------"${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties;;stop)echo "---------- Kafka 停止 ------------"${KAFKA_HOME}/bin/kafka-server-stop.sh;;restart)$0 stop$0 start;;status)echo "---------- Kafka 状态 ------------"count=$(ps -ef | grep kafka | egrep -cv "grep|$$")if [ "$count" -eq 0 ];thenecho "kafka is not running"elseecho "kafka is running"fi;;*)echo "Usage: $0 {start|stop|restart|status}"esac
4.1.6设置开机自启
chmod +x /etc/init.d/kafkachkconfig --add kafka
4.1.7分别启动 Kafka
service kafka start
4.1.8Kafka 命令行操作
@创建topic
kafka-topics.sh --create --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181 --replication-factor 2 --partitions 3 --topic test --zookeeper:定义 zookeeper 集群服务器地址,如果有多个 IP 地址使用逗号分割,一般使用一个 IP 即可--replication-factor:定义分区副本数,1 代表单副本,建议为 2--partitions:定义分区数--topic:定义 topic 名称
@查看当前服务器中的所有 topic
kafka-topics.sh --list --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181
@查看某个 topic 的详情
kafka-topics.sh --describe --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181
@发布消息
kafka-console-producer.sh --broker-list 192.168.10.10:9092,192.168.10.20:9092,192.168.10.30:9092 --topic test
@消费消息
kafka-console-consumer.sh --bootstrap-server 192.168.10.10:9092,192.168.10.20:9092,192.168.10.30:9092 --topic test --from-beginning --from-beginning:会把主题中以往所有的数据都读取出来
@修改分区数
kafka-topics.sh --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181 --alter --topic test --partitions 6
@删除 topic
kafka-topics.sh --delete --zookeeper 192.168.10.10:2181,192.168.10.20:2181,192.168.10.30:2181 --topic test
4.2实例操作:部署zookeeper + kafka 集群
4.2.1安装zookeeper集群
详见本篇博客上文,接着上面的实验继续做(在所有集群服务器操作)
4.2.2下载安装包并安装kafka
4.2.3修改配置文件
4.2.4修改环境变量
4.2.5配置 kafka 启动脚本并设置开机自启,启动kafka
4.2.6Kafka 命令行操作
@创建topic并查看
@发布消息并读取消息
@修改分区数
@删除 topic
五、部署 Filebeat+Kafka+ELK
5.1部署 Filebeat+Kafka+ELK的操作步骤
5.1.1部署 Zookeeper+Kafka 集群
见上文,接着上面的实验做的
5.1.2部署 Filebeat
要搭建ELK,详见之前的博客
安装httpd:yum -y install httpd systemctl start httpd
cd /usr/local/filebeatvim filebeat.ymlfilebeat.prospectors:- type: log enabled: true paths: - /var/log/httpd/access_log tags: ["access"] - type: log enabled: true paths: - /var/log/httpd/error_log tags: ["error"] ......#添加输出到 Kafka 的配置output.kafka: enabled: true hosts: ["192.168.10.10:9092","192.168.10.20:9092","192.168.10.30:9092"] #指定 Kafka 集群配置 topic: "httpd" #指定 Kafka 的 topic #启动 filebeat./filebeat -e -c filebeat.yml
5.1.3部署 ELK,在 Logstash 组件所在节点上新建一个 Logstash 配置文件
vim kafka.confinput { kafka { bootstrap_servers => "192.168.10.10:9092,192.168.10.20:9092,192.168.10.30:9092" #kafka集群地址 topics => "httpd" #拉取的kafka的指定topic type => "httpd_kafka" #指定 type 字段 codec => "json" #解析json格式的日志数据 auto_offset_reset => "latest" #拉取最近数据,earliest为从头开始拉取 decorate_events => true #传递给elasticsearch的数据额外增加kafka的属性数据 }}output { if "access" in [tags] { elasticsearch { hosts => ["192.168.10.40:9200"] index => "httpd_access-%{+YYYY.MM.dd}" } } if "error" in [tags] { elasticsearch { hosts => ["192.168.10.40:9200"] index => "httpd_error-%{+YYYY.MM.dd}" } } stdout { codec => rubydebug }}#启动 logstashlogstash -f kafka.conf
5.1.4启动 logstash
logstash -f filebeat.conf
5.1.5浏览器访问测试
浏览器访问 http://192.168.10.40:5601 登录 Kibana,单击“Create Index Pattern”按钮添加索引“httpd_access-*”,单击 “create” 按钮创建,单击 “Discover” 按钮可查看图表信息及日志信息。
基于zookeeper的kafka中间件
全球快讯:redis缓存穿透、缓存雪崩、缓存击穿【项目总结】
今日热议:零基础开启元宇宙|抖音快手虚拟形象直播【源码】
全球最资讯丨概率论之重要的随机变量
Zabbix技术分享——zabbix命令详解
索尼被曝测试新版PS5:性能不变、增设可拆卸光驱
全球热消息:因担心投资者被吸引到他国:欧盟再次推迟将锂列为有害物质
世界今亮点!苹果担心的事儿发生了 男子用AirTag追踪妻子被逮捕
《巫师3》次时代版体积惊人:PS5升级包超50GB
【世界播资讯】国家邮政局官方回应“快递不快”:加快恢复 邮政顺丰京东开启夜派
环球简讯:eval和ast.literal_eval区别
即时:MySQL索引必知必会
焦点速看:深入理解Whitelabel Error Page底层源码
短讯!字符串函数
【脚本项目源码】Python实现鲁迅名言查询系统
短讯!NGINX常见的变量说明
程序员也可以很浪漫,精选10个圣诞节特效及源码
天天资讯:使用Cpolar搭建一个图片网站2 (Piwigo网站环境准备及安装)
苹果笔记本是什么系统?苹果笔记本怎么安装win10系统?
关闭笔记本触摸板的方法有哪些?关闭笔记本触摸板的四种方法
柳叶刀是哪个国家的杂志?柳叶刀杂志是什么水平?
2023年元旦股市放假几天?2023年元旦股市休市时间表
消防车也要去加油站加油吗?消防车多少钱一台?
win10可以免费升级吗?免费升级win10的条件有哪些?
wlan和wifi哪个网速更快?wlan和wifi的区别是什么?
内存卡读不出来是什么原因?内存卡读不出来怎么修复?
一加8t和一加8pro哪个好?一加8t参数配置
环球微动态丨下载selenium及其适配谷歌浏览器插件chromedriver(含chrome各版本及下载地址)
19.13备库duplicate恢复新主库(二)
iPhone6多少钱?iphone6系统最高可以升级到多少?
锵锵三人行停播原因是什么?锵锵三人行女嘉宾名单大全
沙海吴邪的计划是什么?沙海吴邪的计划成功了吗?
什么边野草花什么口夕阳斜?什么边野草花什么口夕阳斜是什么诗?
九月青豆角产自什么村?九月青豆角怎么冻好吃?
天国的阶梯结局是什么?天国的阶梯演员表
天天热点!什么是Docker容器?(全面了解使用)
环球新资讯:【collection】4.java容器之LinkedList,Stack,CopyOnWriteArrayList
教你用JavaScript实现实时字符计数器
杭州联合银行 x 袋鼠云:打造智能标签体系,助力银行大零售业务转型
天天看点:卡梅隆感染新冠 缺席洛杉矶首映式
想“白嫖”20万的私人飞机 推特被告了
【报资讯】短了5厘米照样强大 迷你SSD硬盘雄起 速度冲向5GB/s
全球聚焦:小米13、iPhone 14 Pro全角度对比:小米正面碾压式完胜 背面有争议
环球快消息!世界杯半决赛现场将播放两首中文歌:你肯定都听过
当前热讯:CSS实现打字机动画效果
环球热消息:行为管理(锐捷交换篇)
环球播报:硬刚小米13!moto X40核心配置官方全面揭晓
环球热头条丨男子帮摔倒大爷报警反被讹引热议:已和解 对方赔偿男子2000元误工费
环球即时:旗舰猛兽!小米万兆路由器明天首销:1799元
天天时讯:太强了!卢伟冰称小米13 Pro应该叫“小米13 Pro Ultra”
【速看料】《三体》动画爆火!两部真人版剧集明年开播:网飞、腾讯对标
环球热推荐:又一合资倒下?家喻户晓的斯柯达或将退出中国市场
环球聚焦:历史首次!英国女孩通过基因编辑治好白血病对抗癌症:生物工程壮举
全球快资讯:框架第三课---作业讲解(数据增删改查),django请求生命周期流程图,django路由层,反向解析
天天热点!第一百一十二篇: JS数组Array(一)数组基本用法
焦点快播:SpringBoot+VUE
Dockerfile指令与Docker-compose容器编排-搭建docker私有仓库
【环球报资讯】伴娘穿露肩礼服肌肉健硕画面太美不敢看 网友:真金刚芭比
IGN给RX 7900 XTX显卡打7分:AMD性价比绝对YES
一加显示器X27发布:2K 165Hz屏、65W PD输出
观天下!小米13深度体验:补上最后一块短板!冲击高端真的要成了?
拒绝向SSD认输!机械硬盘每GB单价已暴跌87%:将越来越便宜
环球微动态丨python中的高阶函数
今日讯!雷军宣布小米13系列将很快登陆全球市场:国外用户激动坏了
【世界新要闻】《三体》动画播放量破1.3亿:豆瓣出现N多1星剧评
每日关注!奇葩!男子恋爱转账11万分手能向女方要回不 法院判决
动态焦点:AMD悄悄把RX 7900 XT的功耗提高了:游戏性能实测公布
世界热点评!FreeSWITCH学习笔记:模块
天天快播:服!敬业新郎一边结婚一边拉业务:39元套餐送1200分钟通话、90G流量和宽带
世界要闻:三亚游客乘观光直升机突遇意外迫降海面:发动机断油失效
视点!秒懂:JCTool 的 Mpsc 超高性能无锁队列 (史上最全+10W字长文)
全球要闻:免费可商用!荣耀HONOR Sans字体来了 附下载
0.89元/片:KN95口罩30片26.9元大促
全球热文:全球首款!大上科技墨水屏显示器上架:25.3寸超大屏幕
天天观速讯丨我们为什么使用Docker
天天滚动:win11上同时安装多个版本的python
当前信息:999元 小米Sound Pro太顶了:用户体验后感觉地板都在共振
天天看点:裸奔还是不行 Win10/11系统依然需要安全软件:免费的就行
每日速讯:顺丰突然又崩了!官方回应:系统异常 已恢复
系列最轻巧流畅系统!MIUI 14开发版首批推送来了
每日讯息!第一章作业
【全球报资讯】架构到底是指什么?
实验七-缓冲区溢出
全球看点:分析师称马斯克暴砍80%服务器订单:供应商Intel很受伤
新能源车换电池多少钱?比亚迪汉8万多 特斯拉13万
世界即时看!山东女生送闺蜜化妆品到内蒙被冻炸 当事人:心里很难过
全球首款真Hi-Fi无线耳机!vivo TWS 3 Pro图赏
2022最强电影!《阿凡达2》北京首映礼今晚举办:提前感受科幻巨作
PopClip使用教程图文详解 2022.12亲测有效
焦点快报!Html+CSS小案例项目:CSS开发小米商城电商产品展示效果
通过命令上传到GitHub
当前播报:对于async和await的使用方式、作用效果不怎么理解 ?没关系,初步看这篇就够了。
记录--记一次前端CSS升级
世界观焦点:中国生物最新研究!灭活疫苗对新冠康复者同样具有保护效力
环球热点评!女子住1楼质问物业为啥要交电梯费 网友力挺
OPPO Find N2 Flip亮相:副屏太惊艳
环球新资讯:日本公布2022年度热门汉字:“战”二次当选 理由奇葩
李诞入手小米13和13 Pro:大玩谐音梗
每日快报!【脚本项目源码】Python制作多功能音乐播放器,打造专属你的音乐播放器
每日关注!雷军分享小米13/MIUI 14内置壁纸原图:大批iPhone用户感谢