zookeeper集群+kafka集群

kafka 3.4.1 在命令中已经不需要zookeeper

zookeeper是一个开源的，分布式的，为了分布式架构提供协调服务的apache的项目。作用：保存元数据

zookeeper的工作机制：

观察者模式设计的分布式服务器管理架构。

负责存储和管理元数据，记录集群的变化。保存集群变化的信息。

zookeeper的特点：

1、在集群中分为领导者和追随者组成的集群。

2、只要有半数以上的节点正常工作，整个zookeeper就可以正常工作，zookeeper集群在部署时一般选择奇数台。

3、全局的数据一致，每个zookeeper不论是领导者还是追随者，在访问他们的数据时都是一致的。

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

5、数据更新的实时性。

6、领导者追随者根据投票产生。

选举机制：

A B C

1、服务器A先启动，发起一次选举，A会投自己一票，A有一票，不够半数，选举无法完成，A进入looking状态。

2、服务器B启动，再发起一次选举，服务器B也投自己一票，服务器A和B用myid做比较，谁的myid大，如果A比B小，A会把票投给B，2票，B自动当选为leader

3、C启动，自动成为追随者，A也是追随者

安装zookeeper

先安装java的依赖环境

安装包解压

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

 进zoo.cfg配置文件中

第二行
tickTime=2000
#服务端与客户端之间心跳线时间，2秒检测一次服务器和客户端之间的通信。
initLimit=10
#领导者和追随者之间。初始连接时能够容忍的超时时间。10：10*tickTime=2000ms=10*2S=20S
syncLimit=5
#同步超时时间。领导者和追随者之间，同步通信超时的时间。5：5*2s=10s。如果超过这个时间。leader会认为follower丢失，会把他移除集群
dataDir=/opt/zookeeper/data
#保存数据的目录，需要单独创建
clientPort=2181
#提供的对外访问端口2181
在16行插入
dataLogDir=/opt/zookeeper/logs
#创建日志文件位置
在最后一行插入
server.1=192.168.124.50:3188:3288
#1：每个zookeeper集群的初始myid
#3188：领导者和追随者之间交换信息的端口(内部通信的端口)
#3288：一旦leader丢失响应，开启选举，3288就是用来执行选举时的服务器之间通信端口。
server.2=192.168.124.51:3188:3288
server.3=192.168.124.60:3188:3288
cd zookeeper
mkdir data logs
#创建在配置文件中定义的两个目录

创建zookeeper的安装路径的目录、日志目录，以及创建myid值，以及给zookeeper给他写一个启动脚本，执行权限，添加到系统服务启动当中

vim /etc/init.d/zookeeper
#!/bin/bash
#chkconfig:2345 20 90
#description:Zookeeper Service Control Script
ZK_HOME='/opt/zookeeper'
case $1 in
start)
    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

启动zookeeper以及产看他的状态的命令

kafka概述：

消息队列：MQ

在高并发环境下，同步的请求来不及处理，请求太多会造成阻塞。

例如大量的请求并发到数据库， too many connection报错

消息队列，使用异步处理方式，可以缓解系统处理请求的压力。

主要就是生产者和消费者

kafka的作用：

1、异步处理

2、系统解耦：每个系统之间独立运行，互相之间没有必然的依赖关系。

（重要）微服务架构中的通信对于解耦至关重要

各个微服务之间独立运行，分别处理各自的请求和消息，就可以提高整个系统的吞吐量和处理能力，尤其是电商的订单系统，网站的工单系统，典型的一个消息队列场景。

3、负载均衡

消息队列的负载均衡：把任务发送到多个消费者，多个消费者可以并行处理队列中的消息。

4、流量控制和限流

通过延迟的方法，处理生产速率和消费者的处理速度（代码控制）

5、数据同步和分发

可以实现跨系统的数据同步和日志收集

6、任务调度和定时任务

7、实时数据处理

8、备份和恢复

消息队列的模式：

1、点对点 一对一消费者消费完数据之后，生产者会自动清除已消费的数据

一个生产者对应一个消费者（淘汰）

2、发布订阅模式（一对多，观察者模式，消费者数据在消费完之后不会被清除，保留一段时间）

生产者发布一个消息，可以是一个消费者使用，也可以是多个消费者同时使用（主流）

kafka就是发布订阅模式的消息队列 RAbbitMQ也是发布/订阅模式的消息队列，小集群内部使用。

大数据的实时处理领域

kafka的特性：

高吞吐量，低延迟

每秒可以处理几十万条数据，而延迟只有几毫秒

集群的可扩展性（热扩展）

消息的持久化：生产者发布的消息可以保存到磁盘当中，防止数据丢失（有时间限制）

容错性：挂了一个可以继续使用

高并发：数千个客户端可以同时读写

面试必问：

kafka的组件：

tpoic 主题，kafka的基本单元，所有生产者发布的消息都是发到主题。

消费者订阅主题，然后消费生产者发布的消息

生产者：生产者把消息发布到主题

消费者：订阅主题，消费生产者发布的消息

分区：每个主题都可以分成多个分区，每个分区都是一个有序的子集

分区当中保留数据，按照偏移量来有序的存储数据，消费者可以根据偏移量来消费指定分区当中的消息（一般不用）

分区还有备份的作用，我们在创建主题的时候创建分区，创建分区时要指定副本数

分区和执行的集群机器数量一般是保持一致的

副本：备份分区的消息，最少要两个，互为备份

偏移量：消息在分区当中的唯一标识，跟踪和定位消息所在的位置，消费者可以根据偏移量来处理信息

经纪人 broker 经纪人处理生产者和消费者的请求（kafka）

元数据（zookeeper）

kafka的工作流程：（重要）

生产者将消息发布到指定的主题，每个消息都附带一个key和value

主题是有多个分区的，生产者把消息写入一个分区(带偏移量)

经纪人（kafka）分配和处理生产者的发布请求，偏移量也是经纪人分配（在分区中是唯一的）

消费者订阅主题，获取全量的消费者的消费信息（默认模式），也可以从指定的分区获取消息（代码来完成，一般不用）

生产者发布的消息会在本地保留一段时间，防止消费者有延迟或者处理速度过慢，导致没有成功消费，保留时间为7天

安装kafka

解压kafka的安装包

 切换进入kafka的config目录中，找到kafka的主配置文件server.properti

21行
broker.id=1
#三台机器的id不能重复
28行
#声明监听端口和id。如果声明了broker.id，这一行可以默认不动
42行
num.network.threads=3
#处理网络请求的线程数量，默认即可
46行
num.io.threads=8
#处理磁盘的io线程数量，一定要比硬盘数大。默认即可
50行
socket.send.buffer.bytes=102400
#发送套接字的缓冲区大小。默认即可
54行
socket.receive.buffer.bytes=102400
#接收者的接受套接字缓冲区大小。默认即可
58行
socket.reques.max.bytes=104657600
#请求套接字的缓冲区大小。单位是字节
65行
log.idrs=/opt/kafka
#指定日志路径
70行
num.partitions=1
#在此Kafka服务器上创建topic，如果不指定默认分区数。默认是1个如果指定了这个配置无效
75行
num.recovery.threads.per.data.dir=1
#用于恢复、回收、清理data下的数据的线程数量。Kafka默认不允许删除主题
110行
log.retention.hours=168
#生产者发布的数据文件在主题当中保存的时间
#168：单位是小时。默认是7天
130行
zookeeper.connect=192.168.124.50:2181,192.168.124.51:2181,192.168.124.60:2181
#配置连接zookeeper集群
其他两台机器修改id后
到65行修改日志信息
log.idrs=/opt/kafka
到123行修改即可
zookeeper.connect=192.168.124.50:2181,192.168.124.51:2181,192.168.124.60:2181

声明一下他的变量路径，在最后一行添加。

vim /etc/profile

export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin

 生效命令

 source /etc/profile

 给kafka写一个启动脚本

#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/opt/kafka'
case $1 in
start)
    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 ];then
        echo "kafka is not running"
    else
        echo "kafka is running"
    fi
;;
*)
    echo "Usage: $0 {start|stop|restart|status}"
esac

 给kafka执行权限以及添加到系统服务启动配置中，然后开启kafka，查看一下端口

创建主题命令

kafka-topics.sh --create --bootstrap-server 192.168.124.50:9092,192.168.124.51:9092,192.168.124.60:9092 --replication-factor 2 --partitions 3 --topic test1
 

--replication-factor 2
创建分区的副本数2
--partitions3 分区数
--topic test1 指定主题的名称

 查看当前服务中的topic主题

 kafka-topics.sh --list --bootstrap-server 192.168.124.50:9092,192.168.124.51:9092,192.168.124.60:9092

创建生产者发布和订阅消息命令

kafka-console-producer.sh --broker-list 192.168.124.50:9092,192.168.124.51:9092,192.168.124.60:9092 --topic test1

 创建消费者消费消息的命令

 kafka-console-consumer.sh --bootstrap-server 192.168.124.50:9092,192.168.124.51:9092,192.168.124.60:9092 --topic test1 --from-beginning

从头开始消费

 查看主题详细信息的命令

kafka-topics.sh --describe --bootstrap-server 192.168.124.50:9092,192.168.124.51:9092,192.168.124.60:9092

 Topic: test1    TopicId: GFCLvfhBSP2EI8NwiNrT1A PartitionCount: 3       ReplicationFactor: 2   Configs:
        Topic: test1    Partition: 0    Leader: 0       Replicas: 0,2   Isr: 0,2
        Topic: test1    Partition: 1    Leader: 2       Replicas: 2,1   Isr: 2,1
        Topic: test1    Partition: 2    Leader: 1       Replicas: 1,0   Isr: 1,0

topic: 主题名称
Partition:分区 偏移量
leader:分区的领导者，用来处理分区的读写操作。只有在指定写分区和指定读分区时才工作。如果不是指定，全量展示，无意义。
Replicas:副本   0    1     2--------broker id
ISR:表示当前于领导者同步的副本    0  1   2   

 删除主题命令

kafka-topics.sh --delete --bootstrap-server  192.168.124.50:9092,192.168.124.51:9092,192.168.124.60:9092    --topic test1

面试题：

kafka的消息堆积该如何解决:
消费者出现了延迟或者处理能力太差，导致消息堆积。
1、减少kafka持久化的保存时间。
2、修改主题的分区数，扩大分区的数量，提高消费者获取的通道
3、可以指定多个消费者共同工作，处理消息的积压。

扩大kafka分区的命令

kafka-topics.sh --bootstrap-server 192.168.124.50:9092,192.168.124.51:9092,192.168.124.60:9092 --alter --topic testl --partitions  +数值           

总结：

zookeeper就是保存集群的元数据

Kafka 是一个分布式的流处理平台，具有以下重要特点和优势：

1. 高吞吐量和低延迟

   • 能够处理每秒数十万的消息，满足大规模数据处理的需求。
   • 例如，在电商网站的订单处理中，快速处理大量的订单创建和更新消息。

2. 分布式架构

   • 数据分布在多个节点上，具有良好的容错性和扩展性。
   • 当一个节点出现故障，其他节点可以继续工作，不影响整体服务。

3. 消息持久化

   • 保证消息不会丢失，即使系统出现故障。
   • 对于金融交易等对数据可靠性要求高的场景至关重要。

4. 解耦生产者和消费者

   • 生产者和消费者可以独立工作，互不影响。
   • 比如，生产者可以持续发送消息，而消费者可以按照自己的节奏处理。

5. 支持多种消息传递模式

   • 包括点对点和发布/订阅模式。

6. 分区机制

   • 可以将主题分为多个分区，实现并行处理和负载均衡。

7. 易于与其他系统集成

   • 能够与 Hadoop、Spark 等大数据处理框架结合使用。

总之，Kafka 在大数据处理、实时数据传输、流处理等领域有着广泛的应用，为构建高性能、可靠的分布式系统提供了强大的支持。