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 先启动，发起第一次选举，投票投给自己，只有一票，不满半数，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 A0 B0 C3

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

D myid4

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

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

1、初始化的情况产生选举

2、服务器之间喝leader丢失了连接状态

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

leader不存在，leader不存在

1、服务器id大的胜出

2、epoch大，直接胜出

3、epoch相同，事务id大的胜出

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

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

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

zookeeper实验：

三台同步操作：

关闭防火墙和安全机制

cd 到opt

然后拖软件包

升级java环境

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

然后一块解压

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

文件改名

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

配置文件改名

cp zoo_sample.cfg zoo.cfg

改配置文件（12行改一下，并且在下面插入一行，在最后一行插入三行）

server.1=192.168.233.10:3188:3288

1每个zookeeper集群的初始myid

192.168.233.10：服务器的ip地址

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

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

创建data和logs

cd data 每个主机分别

echo 1

echo 1 > /opt/zookeeper/data/myid

echo 2

echo 2 > /opt/zookeeper/data/myid

echo 3

echo 3 > /opt/zookeeper/data/myid

然后做一个启动配置文件

vim /etc/init.d/zookeeper

给配置的脚本一个执行权限

chmod +x /etc/init.d/zookeeper

分别加入系统服务

三台服务器分别启动

查看状态（leader有可能在2有可能在3）

消息队列：kafka

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

比方说数据库就会形成行锁或者表锁。请求线程满了，超标了，too many connection整个系统雪崩

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

解耦：

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

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

A ?B ?C

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

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

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

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

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

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

消息队列的模式：

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

RABBITMQ（作业）

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

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

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

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

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

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

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

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

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

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

zookeeper自建一个注意：_consumer_offsets

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

kafka的工作流程：

kafka（2.7.0版本）的实验：

kafka（旧版需要在zookeeper的基础上，新版不用）

三台机子同步操作：

解压kafka

tar zxvf kafka_2.13-2.7.0.tgz

改名

改配置文件

在最后一行添加两行

生效

cd进来

备份配置文件

改配置文件

10：

130行修改一下

168是小时，默认七天

20：

id改成2

30：

三台主机同步操作：

文件

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的执行命令都在/opt/kafka/bin目录下

20：

创建主题：

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

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

3、--replication-factor 2 指的是定义每个分区的副本数

4、partitions 3 指定主题的分区数

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

10：

Partition：分区编号

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

在上述输出中，领导者的编号分别为 3、1、3。

Replicas：每个分区可以有多个副本（Replicas），用于提供冗余和容错性。

在上述输出中，Replica 3、1、2 分别对应不同的 Kafka broker。

Isr：ISR（In-Sync Replicas）表示当前与领导者保持同步的副本。

ISR 3、1分别表示与领导者同步的副本。

三台主机同步操作：

20：

30：

kafka（3.4.1新版本）实验

三台主机同步操作：

关闭防火墙和安全机制

cd到opt目录下

将kafka3.4.1的软件包和zookeeper的软件包拖进来

解压zookeeper组件

安装java环境

zookeeper的组件改名

改名字

改zookeeper的配置文件

在最后一行添加

创建目录

mkdir /opt/zookeeper/data

mkdir /opt/zookeeper/logs

取消同步操作

10：

20：

30：

同步操作：

脚本：

#!/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

查看当前状态

切换到opt目录下

解压kafka

改名kafka

切换到config

做备份

改配置文件

关闭同步操作

10：

34行

20：

id改成2

30：

和上面一样

同步操作：

最后一行添加

配置 Zookeeper 启动脚本

#!/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

10：

创建topic

查看主题topic

查看所有topic的详情

发布消息

30：

20：

实时获取消息

30：

总结：

1、zookeeper：主要是分布式，观察者模式，统一各个服务器节点的数据

在kafka当中，收集保存kafka的元数据

2、kafka消息队列，订阅发布模式

???RABBIT MQ（实现rabbit MQ消息队列）

elk+kafka+filebeat（httpd mysql）(要和上面kafka实验一块使用)实验：

（六台机子）

远程收集两台：httpd mysql

httpd1 mysql1 topic 2.7 3.0

elk1（有logstash）:

切换到opt目录下

把filebeat软件包拖进来

解压软件包

tar -xf filebeat-6.7.2-linux-x86_64.tar.gz

改名

mv filebeat-6.7.2-linux-x86_64 filebeat

切换到filebeat

安装nginx

启动nginx

浏览器访问一下试试看

文件做备份

改配置文件

注释output

再开一台终端给权限

启动filebeat

看日志

再开一台elk1的终端

创建一个文件

input {

????kafka {

????????bootstrap_servers => "20.0.0.10:9092,20.0.0.20:9092,20.0.0.30:9092"

????????topics ?=> "nginx"

????????type => "nginx_kafka"

????????codec => "json"

????????????????auto_offset_reset => "earliest"

????????????????decorate_events => true

????}

}

output {

??if "nginx" in [tags] {

????elasticsearch {

??????hosts => ["20.0.0.20:9200","20.0.0.30:9200"]

??????index => "nginx_access-%{+YYYY.MM.dd}"

????}

??}

??stdout { codec => rubydebug }

}

elk2:

先查看端口9100，如果没启动就在命令行敲

cd elasticsearch-head-master/

npm run start &

elk3:

先查看端口9100，如果没启动就在命令行敲

cd elasticsearch-head-master/

npm run start &