zookeeper集群+kafka集群:

kafka3.0之前依赖于zookeeper。
zookeeper开源，分布式的架构。提供协调服务（Apache项目）
基于观察者模式涉及的分布式服务管理架构。
存储和管理数据。分布式节点上的服务接受观察者的注册。一旦分布式节点上的数据发生变化，由zookeeper开负责通知分布式节点上的服务。

zookeeper：分别为领导者(leader)，追随者 (follower)组成的集群
1、只要有一半以上的集群存活，zookeeper集群就可以正常工作。适用于安装奇数台的服务集群
2、全局数据一致，每个节点都保存相同的数据。维护监控服务的数据一致。
3、数据更新的原子性，要么成功，要么失败
4、实时性，只要有你变化，就立刻同步。

zookeeper的应用场景

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

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

三台服务器 A B C 
A 先启动，发起第一次选举，投票投给自己，只有一票，A的状态是looking
B 启动，再发起一次选举，A和B分别投自己一票，交换选票信息，myid-A发现myid-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
此时结构已确定，后续加的服务器都是追随者。

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

1、初始化的情况会产生选举
2、服务器之间和leader丢失了连接状态
若leader已经存在，建立连接即可
若leader不存在，leader不存在。
则重新选举
1、myid大的胜出
2、EPOCH大的，自己胜出
3、EPOCH相同，事务ID大的胜出。
注：EPOCH 每个eader任期的代号，没有leader，大家的逻辑地址相同，每投完成一次之后，数据递增。
事务id,表示服务器的每一次变更，没变更一次事务iD变化一次。
服务器ID，zookeeper集群当中，每个机器都有一个ID,每台机器不重复，myid保持一致。

实验

zookeeper+kafka(2.7.0)
192.168.10.100    zookeeper+kafka
192.168.10.101    zookeeper+kafka
192.168.10.102    zookeeper+kafka

安装zookeeper

yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel
tar -xf apache-zookeeper-3.5.7-bin.tar.gz
mv apache-zookeeper-3.5.7-bin zookeeper
cd zookeeper/conf/
cp zoo_sample.cfg zoo.cfg
 
vim zoo.cfg
*****************************************************************************************
 
2
tickTime=2000
#服务器与客户端之间心跳时间，2秒检测一次服务器和客户端之间的通信
6
initLimit=10
#领导者和追随者之间，初始连接时容忍的时间 10*2S 20S
10
syncLimit=5
同步超时时间，领导者和追随者之间，同步通信超时的时间，5*2=10s，leader会认为follower丢失，移除集群。
15
dataDir=/opt/zookeeper/data
#保存数据的目录，需要单独创建
dataLogDir=/opt/zookeeper/logs
19
clientPort=2181
34
server.1=192.168.10.100:3188:3288
server.2=192.168.10.101:3188:3288
server.3=192.168.10.102:3188:3288
注
server.X=192.168.10.xxx:3188:3288
X:每个zookeeper集群的初始myid。
192.168.10.xxx 服务器的IP地址
3188 领导者和追随者之间交换信息的端口(内部通信的端口)
3288 一旦leader丢失响应，开启选举，3288就是用来执行选举时的服务器之问通信端口。2
tickTime=2000
#服务器与客户端之间心跳时间，2秒检测一次服务器和客户端之间的通信
6
initLimit=10
#领导者和追随者之间，初始连接时容忍的时间 10*2S 20S
10
syncLimit=5
同步超时时间，领导者和追随者之间，同步通信超时的时间，5*2=10s，leader会认为follower丢失，移除集群。
15
dataDir=/opt/zookeeper/data
#保存数据的目录，需要单独创建
dataLogDir=/opt/zookeeper/logs
19
clientPort=2181
34
server.1=192.168.10.100:3188:3288
server.2=192.168.10.101:3188:3288
server.3=192.168.10.102:3188:3288
注
server.X=192.168.10.xxx:3188:3288
X:每个zookeeper集群的初始myid。
192.168.10.xxx 服务器的IP地址
3188 领导者和追随者之间交换信息的端口(内部通信的端口)
3288 一旦leader丢失响应，开启选举，3288就是用来执行选举时的服务器之问通信端口。
 
 
*****************************************************************************************

//在每个节点上创建数据目录和日志目录
mkdir /opt/zookeeper/data
mkdir /opt/zookeeper/logs
 
cd /opt/zookeeper/data/
//在每个节点的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

为什么要引入消息队列(MQ)，处于服务端和客户端的中间件。在高并发环境下，同步请求来不及处理。来不及处理的请求会形成阻塞。
比方说数据库就会形成行锁或者表锁。请求线程满了，超标了，too many connection

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

耦合：组件与组件之间的的依赖性关系，修改其中一个，对其他组件影响不大，无需修改所有
解耦:A B C 只要通信保证，其他的修改不影响整个集群，每个组件可以的独立的扩展，修改，降低组件之间的依赖性。

可恢复性：系统当中的有一部分组件消失，不影响整个系统。在队列中，即使有一个处理消息的进程失败，一旦恢复，还可以重新加入队列当中，进行处理消息。
缓冲：可以控制和优化数据经过系统的时间和速度。解决生产消息和消费消息处理速度不一致的问题。
峰值的处理能力：消息队列在峰值的情况之下，能够顶住突发的访问压力。避免专门为了突发情况而对系统进行修改。

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

消息队列的模式：

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

发布/订阅模式：一对多，又叫观察者模式。消费者提取数据之后，队列当中的消息不会被清除。
生产者发布一个消息到对象（主题），所有消费者都通过主题获取信息
主题：topic topic类似于一个数据流管道，生产者可以把消息发布到主题，消费者可以从主题中订阅数据。主题可以分区，每个分区都有自己的偏移量。
分区: partition 每个主题都可以分成多个分区。每个分区是数据的有序子集，分区可以允许kafka进行水平拓展，以处理大量数据。
消息在分区中可以按偏移量存储，消费者可以独立的读取每个分区的数据。
偏移量：是每个消息在分区中的唯一标识。消费者可以通过便宜量来跟踪获取已读或者未读消息的位置。也可以提交偏移量来记录已处理的消息。

消费类型：
第一种，begin 从头开始，获取所有
第二种，实时获取，我只消费，后续产生的消息
第三种，指定偏移量消费，代码实现

生产者: producer 生产者把数据发送kafka的主题当中，负责写入消息
消费者: 从主题当中读取数据，消费者可以说一个也可以是多个。每个消费者有一个唯一的ID，kafka通过消费者实现负载均衡和负载均衡和容错性
经纪人: Broker每个kafka节点都有一个Broker，每个Broker负责一台kafka服务器，id唯一，存储主题分区当中数据，处理生产和消费者的请求。维护元数据(zookeeper)

zookeeper：负责保存元数据，元数据就是topic的相关信息(发在哪台主机上，指定了多少分区，以及副本数，偏移量。
会自建一个主题：__consumer_offsets
3.0之后不依赖zookeeper的核心，元数据由kafka节点自己管理。

kafka的工作流程

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

kafka安装

tar -xf kafka 2.13-2.7.0.tgz
vim /etc/profile
****************************************************************************
 
export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin
 
****************************************************************************
source /etc/profile
cd /opt/kafka/config
vim server.properties
*****************************************************************************************
 
60
log.dirs=/var/log/kafka
#日志存放路径
zookeeper.connect=192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181
 
-----------------------------------------------------------------------------------------
 
broker.id=1、2、3
42
num.network.threads=3
#处理网络请求的线程数量，默认即可
46
num.io.threads=8
#处理磁盘的IO线程数量，一定要比硬盘数大。
50
socket.send.buffer.bytes=102400
#发送套接字的缓冲去大小
socket.receive.buffer.bytes=102400
#接受者的接受套接字缓冲区大小。
socket.request,max.bytes=104857600
#请求套接字的缓冲区大小。（字节）
 
65
log.dirs=/var/log/kafka
#日志存放路径
70
num.partitions=1
#在此kafka服务器上创建topic，默认分区数。如果指定了，这个配置无效了
75
num.recoverythreads.per.data.dir=1
#用来恢复，回收，清理data下的数据的线程数量。kafka默认不允许删除主题。
    110
log.retention.hours=168
#生产者发布的数据文件在主题当中保存的时间。168小时=7天
130
zookeeper.connect=192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181
 
*****************************************************************************************

vim /etc/init.d/kafka
chmod +x /etc/init.d/kafka
chkconfig --add kafka
service kafka start    #启动Kafka
netstat -antp | grep 9092    #Kafka的默认端口号9092,查看服务端口是否启动

创建主题

1、在kafka的bin日录下，是所有的kafka可执行命令文件
2、--zookeeper 指定的是zookeeper的地址和端口，保存kafka的元数据
3、--replication-factor 2 定义每个分区的副本数
4、partitions 3 指定主融的分区数
5、--topic test1 指定主题的名称
 

kafka-topics.sh --create --zookeeper IP地址:2181 --replication-factor 副本数 --partitions 分区数 --topic 主题名
--replication-factor 副本数  
--partitions 分区数
--topic 主题名  
 
例：
kafka-topics.sh --create --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181 --replication-factor 2 --partitions 3 --topic test1

查看主题

查看所有主题

查看所有的topic
kafka-topics.sh --describe --zookeeper IP地址:2181
例
kafka-topics.sh --describe --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181

查看指定主题

查看指定主题
kafka-topics.sh --describe --zookeeper IP地址:2181 --topic 主题名
例
kafka-topics.sh --describe --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181 --topic test2

查看内容声明

Partition：分区编号	
 
Leader：每个分区都有一个领导者（Leader），领导者负责处理分区的读写操作。
在上述输出中，领导者的编号分别为 3、1、3。
 
Replicas：每个分区可以有多个副本（Replicas），用于提供冗余和容错性。
在上述输出中，Replica 3、1、2 分别对应不同的 Kafka broker。
 
Isr：ISR（In-Sync Replicas）表示当前与领导者保持同步的副本。
ISR 3、1分别表示与领导者同步的副本。

vim /etc/hosts
*******************************************************************************************
 
192.168.10.100 test1
192.168.10.101 test2
192.168.10.102 test3
 
*******************************************************************************************

生产者：

kafka-console-producer.sh --broker-list IP地址:9092 --topic 主题名
例
kafka-console-producer.sh --broker-list 192.168.10.100:9092,192.168.10.101:9092,192.168.10.102:9092 --topic test1

消费者：

同步所有的
kafka-console-consumer.sh --bootstrap-server IP地址:9092 --topic 主题名 --from-beginning
例：
kafka-console-consumer.sh --bootstrap-server 192.168.10.100:9092,192.168.10.101:9092,192.168.10.102:9092 --topic test1 --from-beginning
 
实时同步
kafka-console-consumer.sh --bootstrap-server IP地址:9092 --topic 主题名
例：
kafka-console-consumer.sh --bootstrap-server 192.168.10.100:9092,192.168.10.101:9092,192.168.10.102:9092 --topic test1

修改分区数

格式
kafka-topics.sh --zookeeper ip地址:2181 --alter --topic 主题名 --partitions 分区数
例
kafka-topics.sh --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181 --topic test1 --partitions 3

删除主题

格式
kafka-topics.sh --delete --zookeeper ip地址:2181 --topic 主题名
例:
kafka-topics.sh --delete --zookeeper 192.168.10.100:2181,192.168.10.101:2181,192.168.10.102:2181 --topic test1

总结：

1、zookeeper;主要是分布式，观察者模式，统一各个节点的数据。
在kafka当中，收集保存kafka的元数据
2、kafka消息队列订阅发布模式，速度快，吞吐量高，占用系统资源多
    RABBIT MQ (实现rebbit MQ 消息队列)