1.zookeeper

kafka3.0之前依赖于zookeeper

zookeeper是一个开源，分布式的架构，提供协调服务（Apache项目）

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

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

zookeeper分为领导者，追随者 leader follower组成的集群

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

全局数据一致，每个zookeeper每个节点都保存相同的数据，维护监控服务的数据一致。（主要作用）

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

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

zookeeper的应用场景

1.统一命名服务，在分布式的环境下，对所有的应用和服务进行统一命名

2.统一配置管理，配置文件同步，kafka的配置文件被修改，可以快速同步到其他节点

3.统一集群管理，实时掌握所有节点的状态

4.服务器动态上下限

5.负载均衡，把访问的服务器的数据，发送到访问最少的服务器处理客户端的请求

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

三台服务器：A B C

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

B 启动 在发起一次选举，A和B分别投自己一票，交换选票信息，myid A发现B的myid比A的大，A的这票会转而投给B，A 0 B 2 没有半数以上结果，A B 会进入looking 、B有可能leader

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

C的状态变为leader A B变为follower

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

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

1.初始化的情况会产生选举

2.服务器之间和eader丢失了连接状态

特殊情况下：

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

leader不存在，leader不存在

1.服务器ID大的胜出

2.EPOCH大，直接胜出

3.EPOCH相同，事务ID大的胜出

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

事务id表示服务器的每一次更新，每变更一次id变化一次‘

服务器ID

zookeeper 当中所有机器。每台机器不重复，和mysql保存一直

zookeeper+kafka（2.7.0）

kafka （3.4.1）

实现

zookeeper集群

192.168.233.10 zookeeper+kafka

192.168.233.20 zookeeper+kafka

192.168.233.30 zookeeper+kafka

所有

systemctl stop firewalld

setenforce 0

cd /opt

拖进去 apache-zookeeper-3.5.7-bin.tar.gz kafka_2.13-2.7.0

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

java -version

cd /opt

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

mv apache-zookeeper-3.5.7-bin zookeeper

cd zookeeper

cd /opt/zookeeper/conf

cp zoo_sample.cfg zoo.cfg

vim zoo.cfg

tickTime=2000

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

initLimit=10

#领导者和追随者之间，最对几次心跳数超时 10*2S 20S

syncLimit=5

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

16行

dataDir=/opt/zookeeper/data

dataLogDir=/opt/zookeeper/logs (需要改)

添加

server.1=192.168.233.10:3188:3288

server.2=192.168.233.20:3188:3288

server.3=192.168.233.30:3188:3288

##

server.1=192.168.233.10:3188:3288

1 定义每个zookeeper集群的初始myid

192.168.233.10 ：服务器的ip地址

3188：领导者和追随者之间交换信息的端口（内部通信的端口）

3288：一旦leader丢失响应，开启选举，3288就是用来执行选举时的通信端口

##

wq!

mkdir /opt/zookeeper/data

mkdir /opt/zookeeper/logs

三台分别创建

10：

cd ..

cd data/

echo 1 > myid

20：

cd ..

cd data/

echo 2 > myid

30:

cd ..

cd data/

echo 3 > myid

所有

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

wq!

三台服务器分别启动(一个一个起)

chmod +x /etc/init.d/zookeeper

chkconfig --add zookeeper

service zookeeper start

起完以后看状态

service zookeeper status

2.kafka

为什么要引入消息队列（MQ），首先他也是一个中间件。在高并发环境下，同步请求太多来不及处理，来不及处理的请求会阻塞，比方数据库就会形成行锁或者表锁，请求线程满了，超标了，too many connection。整个系统雪崩

消息队列的作用：异步处理请求 流量削峰，应用解耦 可恢复性 缓冲机制

解耦：

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

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

A B C

只有通信保证，其他的修改不影响整个集群，每个组件可以独立的扩展，修改，降低组件之间的依赖性

依赖点就是接口约束，通过不同的端口，保证集群通信

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

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

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

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

消息队列的模式：

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

RABBITMQ

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

生产者发布一个消息到对象（主题）所有消费者都是通过主题获取消费之后，队列当中的消息不会消除

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

分区：partition 每个主题都可以分成多个分区，每个分区是数据的有序子集，分区可以运行kafka进行水平扩展，以处理大量数据

消息在分钟按照偏移量存储，消费者可以独立每个分区的数据

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

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

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

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

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

zookeeper自建一个主题：_consumer_offsets,

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

消费的方式：

begining，从头开始

实时更新

指定位置，用代码编写

kafka的工作流程：

生产者向主题里面发送数据，主题里的分区保存数据，消费者根据消费方式来消费数据。

生产者写入的topic的数据时持久化，默认7个小时

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

实验

所有

cd /opt

tar -xf kafka_2.13-2.7.0.tgz

mv kafka_2.13-2.7.0 kafka

vim /etc/profile

export KAFKA_HOME=/opt/kafka

export PATH=$PATH:$KAFKA_HOME/bin

source /etc/profile

cd /opt/kafka/config

cp server.properties server.properties.bak

10:

vim server.properties

21行：

65行 修改

130行 修改

zookeeper.connect=192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181

配置zookeeper的集群

20：

所有：

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

wq!

chmod +x /etc/init.d/kafka

chkconfig --add kafka

service kafka start

netstat -antp | grep 9092

所有：

cd /opt/kafka/bin

20:

kafka-topics.sh --create --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181 --replication-factor 2 --partitions 3 --topic test1

kafka-topics.sh --create --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181 --replication-factor 2 --partitions 3 --topic test1

##

创建主题

1.在kafka的bin目录下创建，是所有kafka可执行命令的文件

2. --zookeeper指定的是zookeeper的地址和端口，保存kafka的元数据

3. --replication-factor 2 定义每个分区的副本数

4.partition 3 指定主题的分区数

5. --topic test1 指定主题的名称

10:

kafka-topics.sh --describe --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181

kafka-topics.sh --describe --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181 --topic test1

kafka-topics.sh --describe --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181

kafka-topics.sh --describe --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181 --topic test3

所有

vim /etc/hosts

192.168.233.10 test1

192.168.233.20 test2

192.168.233.30 test3

记得改主机名

20:

kafka-console-producer.sh --broker-list 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3

kafka-console-producer.sh --broker-list 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1

30:

从头开始

kafka-console-consumer.sh --bootstrap-server 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3 --from-beginning

kafka-console-consumer.sh --bootstrap-server 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1 --from-beginning

实时消息

kafka-console-consumer.sh --bootstrap-server 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3

kafka-console-consumer.sh --bootstrap-server 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1

20：创建

kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --partitions 1 --replication-factor 1 --topic guoqi1

kafka-topics.sh --create --zookeeper 192.168.66.16:2181 --partitions 1 --replication-factor 1 --topic guoqi1

30:创建

kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --partitions 1 --replication-factor 1 --topic guoqi2

kafka-topics.sh --create --zookeeper 192.168.66.17:2181 --partitions 1 --replication-factor 1 --topic guoqi2

10:连接

kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic guoqi1

kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic guoqi2

kafka-console-consumer.sh --bootstrap-server 192.168.66.16:9092 --topic guoqi1

kafka-console-consumer.sh --bootstrap-server 192.168.66.17:9092 --topic guoqi2

20:发送

kafka-console-producer.sh --broker-list 192.168.233.20:9092 --topic guoqi1

kafka-console-producer.sh --broker-list 192.168.66.16:9092 --topic guoqi1

30:发送

kafka-console-producer.sh --broker-list 192.168.233.30:9092 --topic guoqi2

kafka-console-producer.sh --broker-list 192.168.66.17:9092 --topic guoqi2

如何修改分区数

20：kafka-topics.sh --zookeeper 192.168.233.20:2181 --alter --topic guoqi1 --partitons 3

删除

kafka-topics.sh --delete --zookeeper 192.168.233.20:2181 --topic guoqi1

查看元数据

cd /opt/zookeeper/bin

./zkCli.sh -server 192.168.233.10:2181