当前位置：首页 > news >正文

2、zookeeper和kafka

news 来源：原创 2025/5/17 19:04:46

zookeeper

zookeeper基本知识

zookeeper：开源的分布式框架协调服务

zookeeper的工作机制

基于观察者模式设计的分布式结构，复制存储和管理架构当中的元信息，架构当中的应用接受观察者的监控，一旦数据有变化，通知对于的zookeeper，保存变化的信息。

zookeeper特点

1、最少要有3台服务器，一个领导者（leader），多个跟随者（follower）。

2、zookeeper要有半数以上的节点存货，整个架构就可以正常工作，所有都是奇数台部署

3、全局数据一致

4、数据更新的原子性，要么都成功，要么都失败。

5、实时性

zookeeper的数据结构

1、统一的命名服务，不是以IP来记录，可以用域名也可以是主机名来记录信息。

2、统一配置管理，所有的节点信息的配置要是一致的。

3、统一的集群管理，在整个分布式的环境中，必须实时的掌握每个节点的状态，如果状态发生变化，要及时更新。

zookeeper安装实操

这里我们使用三台服务器完成zookeeper架构，leader和follow是随机选举的。

实验架构如下：

zw6：192.168.254.16

tomcat：192.168.254.21

mysql：192.168.254.33

1、三台服务器都解压zookeeper包并安装java依赖环境

tar -xf apache-zookeeper-3.5.7-bin.tar.gz

apt -y install openjdk-8-jre-headless

2、三台服务器都创建配置文件zoo.cfg

3、三台服务器修改配置文件zoo.cfg

定义zookeeper的数据目录和日志目录

server.1：指的是服务器的序号
3188：leader和follower之间交换信息的通信端口
3288：选举的端口

4、三台服务器都创建数据目录和日志目录并赋权

5、三台服务器分别在/data/目录下，创建myid文件，并且里面输入分别是1、2、3，和我们配置文件里面1、2、3对应。

6、三台主机都启动zookeeper并查看状态，这里我们mysql3服务器是leader，两外两台都是follow

kafka

消息队列

kafka消息队列：服务端向客户端发送一个指令，客户端收到指令并且通过这个指令反馈到服务端，完成一个异步方式的通信的流程。

kafka消息队列：处理大数据场景非常合适

rabbitMQ消息队列：处理小数据场景合适

activeMQ消息队列：处理小数据场景合适

消息队列的应用场景

1、异步处理：用于用户的短信验证码，邮件通知。

2、系统解耦：用于微服务架构中的服务之间通信。可以降低各个组件之间的依赖程度（耦合度），可以提高组件的灵活性和可维护性。

3、负载均衡：用于高并发系统的任务处理。消息队列把多个任务分发到多个消费者实列，如电商平台的订单系统。

4、流量控制和限流：根据api请求来进行处理。通过控制消息的生产速度和消费者的处理速度来完成限流。

端口：应用和应用之间通信
api接口：应用程序内部各个组件之间的通信的方式

5、任务调度和定时任务：消息队列可以定时的进行任务调度，按照消费者的计划生产对于的数据

6、数据同步和分发：用于日志收集和数据收集。可以远程的实现数据的统一管理。

7、实时数据处理

8、备份和恢复

消息队列的模式

1、点对点：一对一，生产者生产消息，消费者消费消息，这个是一对一的。

2、发布/订阅模式：消息的生产者发布一个主题，其他的消费者订阅这个主题，从而实现一对多。

主题：topic

kafka的组件名称

1、主题 topic：主题是kafka数据流的一个基本的单元，类似于数据的管道，生产者将消息发布到主题，其他消费者订阅主题来消费消息，主题可以被分区，分区有偏移量。

2、生产者：将消息写入主题和分区

3、消费者：从主题和分区中接受发布的消息，一个消费者可以订阅多个主题。

4、分区：一个主题可以被分为多个分区，每个分区都是数据的有序的子集，分区越多，消费者消费的速度越快，可以避免生产者的消费堆积。分区当中有偏移量，按照偏移量进行有序存储，消费者可以独立的读写每个分区的数据。

如何读取分区的数据，一般是不考虑的。只有在消息出现堆积的时候，会考虑扩展分区数。
堆积：消费者没有及时处理掉生产者发布的消息，导致消费堆积

kafka的消费堆积出现如何处理：

1、扩展分区数

2、偏移量：消息在分区当中的唯一标识，根据偏移量指定消费者获取消息的位置。

3、经纪人zookeeper：存储分区的消息，kafka集群的元数据。

kafka数据流向

kafka安装实操

这里我们依然使用上面zookeeper实验的三台服务器完成kafka实操

kafka非常占内存，可以在安装使用前清理下内存，echo 3 > /proc/sys/vm/drop_caches

1、三台服务器解压软件包后移动至/usr/local/kafka

2、三台服务端修改kafka配置文件server.properties

首先配置三台服务器不一样的地方

#全局唯一编号，每一台服务器都不能相同，这里我们定0、1、2即可

#地址分别对应自己主机的IP地址

接下里都是一样的配置

3、三台服务器添加kafka全局变量

4、三台服务器一起启动kafka

./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

看端口可以看出我们kafka服务已启，并且推选了mysql3这个主机为主

kafka补充应用

创建主题

kafka-topics.sh --create --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --replication-factor 2 --partitions 3 --topic test1

bootstrap-server：这里的地址一般是一个集群当中的地址即可，约定是全写。
replication-factor 2：定义主题的副本数，2个副本，一般2、4个，副本是偶数。
partitions 3：定义分区数，一般3、6个

查询当前集群中的主题数

./kafka-topics.sh --list --bootstrap-server 192.168.233.61:9092

发布消息

kafka-console-producer.sh --broker-list 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --topic test1

消费消息

kafka-console-consumer.sh --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --topic test1 --from-beginning

修改主题分区数

查看分区

Leader：每一个分区都有一个Leader，领导者负责处理分区的读写操作

Replicas：副本数，0 1 2 分别对象每一个Leader

Isr：标识和哪个Leader进行同步

Partition：分区的编号

修改分区数

kafka-topics.sh --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --alter --topic test1 --partitions 6

删除主题

kafka-topics.sh --delete --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --topic test1

elk+filebeat+kafka实操

数据流向如下

这里我们结合之前ELK+filebeat实验和上面kafka实验，具体架构如下

es1：zw4，192.168.254.14

es2：zw5，192.168.254.15

logstash、kibana：mysql1，192.168.254.31

filebeat：mysql2，192.168.254.32

kafka：zw6，192.168.254.16

kafka：tomcat，192.168.254.21

kafka：mysql，192.168.254.33

1、修改filebeat配置文件 /usr/local/filebeat/filebeat.yml，更换下标签，区分下之前做的实验，具体修改如下

2、创建logstash的日志收集文件/etc/logstash/conf.d/kafka.sh

codec => "json"：指定数据的格式是json
auto_offser_reset => "latest"：latest，从尾部开始；earliest的话就是从头开始拉取
decorate_events => true：传递给es的数据额外的附加kafka的属性数据

3、启动和验证

启动logstash：logstash -f kafka.conf --path.data /opt/testa &

打开kafka集群的消费集群

kafka-topics.sh --bootstrap-server 192.168.254.16:9092,192.168.254.21:9092,192.168.254.33:9092 --topic nginx_mysql --from-beginning

启动filebeat：./filebeat -e -c filebeat.yml

显示filebeat与kafka集群连接成功

最后我们登录kibana，显示日志收集成功