zookeeper集群+kafka集群

zookeeper集群+kafka集群：

Kafka3.0 之前依赖于zookeeper

Zookeeper开源，分布式的架构。提供协调服务（Apache项目）

基于观察者模式涉及的分布式服务管理架构

存储和管理数据。分布式节点上的服务接受观察者的注册。一旦分布式节点上的数据发生变化，由zookeeper开发负责通知分布式节点上的服务

zookeeper：分为领导者和追随者 ?leader follower组成的集群

只要有一半以上的集群存活，zookeeper集群就可以正常工作。适用于安装奇数台的服务集

群。

全局数据一致，每个zookeeper每个节点都保存相同的数据。维护监控服务的数据——一致

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

实时性，只要有变化，立刻同步。

zookeeper的应用场景：

1. 统一命名服务，在分布式的环境下，对所有的应用和服务进行统一命名。
2. 统一配置管理，配置文件同步，kafka的配置文件被修改，可以快速同步到其他节点。
3. 统一集群管理，实时掌握所有节点的状态。
4. 服务器动态上下线。
5. 负载均衡，把访问的服务器的数据，发送到访问最少的服务器处理客户端的请求。

。

领导者和追随者：zookeeper的选举机制

三台服务器： A B C

A 先启动，发起第一次选举，投票给自己，只有1票，不满半数，A的状态是looking

B 启动，再发起一次选举，A和B分别投自己一票，交换选票信息，myid，A发现B的myid

??比A大，A的这一票转而投给B

A 0 ?B 2 没有半数以上结果，A B会进入looking

C 启动 myid C的myid最大，A和B都会把票投给C

A 0 B 0 C 3 ，C的状态变为leader，A和B变成follower

只要leader确定，后续的服务器都是追随者

只有两种情况会开启选举机制：

1. 初始化的情况会产生选举
2. 服务器之间和leader丢失了连接状态

leader已经存在，建立连接即可

leader不存在

1. 服务器ID大的胜出
2. EPOCH大的直接胜出
3. EPOCH相同，事务ID大的胜出

EPOCH每个leader任期的代号，没有leader，大家的逻辑地址相同，每投完一次之后，数据是递增。

事务ID，表示服务器的每一次变更，每变更一次事务ID变化一次。

服务器ID，zookeeper集群当中的机器都有一个ID，每台机器不重复，和myid保持一致。

准备 3 台服务器

20.0.0.10 ?zookeeper+kafka

20.0.0.11 ?zookeeper+kafka

20.0.0.12 ?zookeeper+kafka

部署 Zookeeper 集群

所有服务器

关闭防火墙

systemctl stop firewalld

systemctl disable firewalld

setenforce 0

安装 JDK

yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel

java -version

安装 Zookeeper

cd /opt

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

mv apache-zookeeper-3.5.7-bin /opt/zookeeper

修改配置文件

cd /opt/zookeeper/conf/

cp zoo_sample.cfg zoo.cfg

vim zoo.cfg

tickTime=2000

#服务器与客户端之间心跳时间，2秒检测一次服务器和客户端之间的通信

initLimit=10

#领导者和追随者之间，初始连接时能够容忍的超时时间。10*2 20s

syncLimit=5

#同步超时时间，领导者和追随者之间，同步通信超时的时间，5*2s，leader会认为follower丢失，移除集群

dataDir=/opt/zookeeper/data

#修改，指定保存Zookeeper中的数据的目录，目录需要单独创建

dataLogDir=/opt/zookeeper/logs

#添加，指定存放日志的目录，目录需要单独创建

clientPort=2181 ??

#客户端连接端口

在最底行添加集群信息

server.1=20.0.0.10:3188:3288

server.2=20.0.0.11:3188:3288

server.3=20.0.0.12:3188:3288

server.1=20.0.0.10:3188:3288

1：每个zookeeper集群的初始myid。20.0.0.10:服务器的ip地址

3188:领导者和追随者之间交换信息的端口(内部通信的端口)

3288:一旦leader丢失响应，开启选举，3288就是用来执行选举时的服务器之间通信端口。

在每个节点上创建数据目录和日志目录

mkdir -p /opt/zookeeper/data

mkdir -p /opt/zookeeper/logs

在每个节点的dataDir指定的目录下创建一个 myid 的文件,不同节点分配1、2、3

echo 1 > /opt/zookeeper/data/myid

echo 2 > /opt/zookeeper/data/myid

echo 3 > /opt/zookeeper/data/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

设置开机自启

chmod +x /etc/init.d/zookeeper

chkconfig --add zookeeper

分别启动 Zookeeper

service zookeeper start

#查看当前状态

service zookeeper status

?Kafka

#安装 Kafka

cd /opt/

tar zxvf kafka_2.13-2.7.0.tgz

mv kafka_2.13-2.7.0 kafka/

#修改配置文件

cd /opt/kafka/config/

vim server.properties

vim server.properties

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=/var/log/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=20.0.0.12:2181,20.0.0.11:2181,20.0.0.12:2181

#配置连接zookeeper集群

其他两台机器修改id后

到65行修改日志信息

log.idrs=/var/log/kafka

到123行修改即可

zookeeper.connect=20.0.0.12:2181,20.0.0.11:2181,20.0.0.12:2181

#修改环境变量日志段是主题分区日志文件的一部分。

vim /etc/profile

export KAFKA_HOME=/opt/kafka

export PATH=$PATH:$KAFKA_HOME/bin

#立即生效

source /etc/profile

三台机器同时操作

创建启动文件脚本

vim /etc/init.d/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

#设置开机自启

chmod +x /etc/init.d/kafka

chkconfig --add kafka

#分别启动 Kafka

service kafka start

#三台主机同步操作主机映射

vim /etc/hosts

20.0.0.10 test1

20.0.0.11 test2

20.0.0.12 test3

#随便选一台服务器创建topic

kafka-topics.sh --create --zookeeper 20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181 --replication-factor 2 --partitions 3 --topic test1

#查看当前服务器中的所有 topic

kafka-topics.sh --list --zookeeper 20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181

#查看某个 topic 的详情

?kafka-topics.sh ?--describe --zookeeper 20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181

#发布消息

?kafka-console-producer.sh --broker-list 20.0.0.10:9092,20.0.0.11:9092,20.0.0.12:9092 ?--topic test1

>1

>2

>3

>4

#消费消息

?kafka-console-consumer.sh--bootstrap-server 20.0.0.10:9092,20.0.0.11:9092,20.0.0.12:9092 ?--topic test1 --from-beginning

>1

>2

>3

>4

--from-beginning：会把主题中以往所有的数据都读取出来

#修改分区数

kafka-topics.sh --zookeeper 20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181?--alter --topic test1 --partitions 6

分区数越多，存储越多，并发量越快。一个分区两个副本即可

#删除 topic

kafka-topics.sh --delete --zookeeper 20.0.0.10:2181,20.0.0.11:2181,20.0.0.12:2181?--topic test1

此时命令执行后，只是打赏打上删除的标记，并没有完全删除。还是保存在元数据当中

为什么要引入消息队列(MQ)，他也是一个中间件，在高并发环境下，同步请求来不及处理。来不及处理的请求会形成阻塞。

例如：数据库就会形成行锁或者表锁

请求线程满了超标了就会出现错误：too many connection

一旦报错too many connection就会引发整个系统雪崩。

此时消息队列的作用就显得至关重要

消息队列的作用：异步处理请求，流量削峰，应用解耦。

耦合：在软件系统当中，修改一个组件需要修改其他所有组件。这就是高度耦合。

低度耦合：修改一个对其他的组件影响不大，无需修改所有

解耦：

例如：A B C

解耦的核心作用就是降低组件之间的依赖性

只要通信保证，其他的修改不影响整个集群，每个组件可以的独立的扩展，修改，降低组件之间的依赖性。依赖点就是接口约束，通过不同的端口，保证集群通信。I

可恢复性:系统当中的有一部分组件消失，不影响整个系统。也就是说在消息队列当中，即使有一个处理消息的进程失败，一旦恢还可以重新加入到队列当中，继续处理消息。

可恢复性:系统当中的有一部分组件消失，不影响整个系统。也就是说在消息队列当中，即使有一个处理消息的进程失败，一旦恢复还可以重新加入到队列当中，继续处理消息。

缓冲:可以控制和优化数据经过系统的时间和速度。解决生产消息和消费消息处理速度不一致的问题。

峰值的处理能力:消息队列在峰值情况之下，能够顶住突发的访问压力。避免专门为了突发情况而对系统进行修改。

异步通信:允许用户把一个消息放入队列，但是不立即处理，等用户想处理的时候在处理。

消息区列的模式:

点对点一对一:消息的生产者发送消息到队列中，消费者从队列中提取消息，消费者提取完之后，队列中被提取的消息将会被移除。后续消费者不能再消费队列当中的消息。消息队列可以有多个消费者，但是一个消息，只能由一个消费者提取。

发布/订阅模式：一对多，又叫做观察者模式。消费者提取数据之后，队列当中的消息不会清除。

生产者发布一个消息到主题，所有消费者都是通过主题获取消息

主题: topic topic类似一个数据流的管道，生产者把消息发布到主题。消费者从主题当中订阅数据。主题可以分区，每个分区都有自己的偏移量。

分区: partition每个主题都可以分成多个分区。每个分区是数据的有序子集，分区可以允许kafka进行水平拓展，以处理大量数据。消息在分钟按照偏移量存储，消费者可以独立读取每个分区的数据。

偏移量:是每个消息在分区中唯一的标识。消费者可以通过便宜量来跟踪获取已读或者未读消息的位置。也可以提交偏移量来记录已处理的信息。

生产者: producer生产者把数据发送kafka的主题当中，负责写入消息。

消费者: consumer从主题当中读取数据.消费者可以是一个也可以是多个。每个消费者有一个唯一的消费者组ID，kafka通过消费者实现负载均衡和容错性。

经纪人：Broker 每个kafka节点都有一个Broker，每个负责一台kafka服务器，id唯一，存储主题分区当中数据，处理生产和消费者的请求。维护元数据（zookeeper）

zookeeper：zookeeper负责保存元数据，元数据就是topic的相关信息（发在哪台主机上，指定了多少分区，以及副本数，偏移量。）

zookeper自建一个注意：_consumer_offsets

3.0之后不依赖zookeeper的核心，元数据由kafka节点自己管理

kafka的工作流程：

1、 生产者向主题topic中发送数据。数据会按照分区依次保存在不同的分区中。分区的偏移量是从0开始

2、 消费者从开始位置消费，只能获取到test1数据

3、 消费者实时消费则会获取到最新的test2数据

4、 此时生产者写入新的数据test3。这时消费者停止消费请求。此时test3会持久化。（生产者写入topic的数据是持久化的，默认是7小时）此时消费者再次开启消费请求，则可以访问到test3。持久化时间过期后则访问不到test3数据。

Kafka也支持延时发送

至少一次语义：只要消费者进入，确保消息至少被消费一次

总结

1、 zookeeper主要是分布式、观察者模式，统一各个服务器节点的数据。在Kafka当中，手机保存Kafka的元数据

2、 Kafka消息队列：是订阅发布模式

Kafka：速度快，资源占用量高

RABBIT MQ：轻量级

3、 Kafka的组件：

主题：topic。一个主题就相当于1个微信群。

分区：处理消息的位置

偏移量：消息产生的时间位置

生产者：产生数据

消费者：消费数据

经纪人：负责管理生产者发布的数据

创建主题一定要有分区，创建分区一定要有副本