Zookeeper集群 + Kafka集群

Zookeeper 概述

Zookeeper 定义
ZooKeeper是一个开源的分布式协调服务，用于构建可靠的分布式系统。它通过提供高度可靠的数据存储和同步机制，来协调不同节点之间的通信和协作。

Zookeeper 工作机制
Zookeeper从设计模式角度来理解：是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。也就是说 Zookeeper = 文件系统 + 通知机制。

Zookeeper 特点
（1）Zookeeper：一个领导者（Leader），多个跟随者（Follower）组成的集群。
（2）Zookeeper集群中只要有半数以上节点存活，Zookeeper集群就能正常服务。所以Zookeeper适合安装奇数台服务器。
（3）全局数据一致：每个Server保存一份相同的数据副本，Client无论连接到哪个Server，数据都是一致的。
（4）更新请求顺序执行，来自同一个Client的更新请求按其发送顺序依次执行，即先进先出。
（5）数据更新原子性，一次数据更新要么成功，要么失败。
（6）实时性，在一定时间范围内，Client能读到最新数据。

Zookeeper 数据结构
ZooKeeper数据模型的结构与Linux文件系统很类似，整体上可以看作是一棵树，每个节点称做一个ZNode。每一个ZNode默认能够存储1MB的数据，每个ZNode都可以通过其路径唯一标识。

部署zookeeper集群

//准备 3 台服务器做 Zookeeper 集群
192.168.80.10
192.168.80.11
192.168.80.12

 安装jdk

//安装 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 /usr/local/zookeeper-3.5.7

修改配置文件

cd /usr/local/zookeeper-3.5.7/conf/
cp zoo_sample.cfg zoo.cfg
 
vim zoo.cfg
tickTime=2000   #通信心跳时间，Zookeeper服务器与客户端心跳时间，单位毫秒
initLimit=10    #Leader和Follower初始连接时能容忍的最多心跳数（tickTime的数量），这里表示为10*2s
syncLimit=5     #Leader和Follower之间同步通信的超时时间，这里表示如果超过5*2s，Leader认为Follwer死掉，并从服务器列表中删除Follwer
dataDir=/usr/local/zookeeper-3.5.7/data      ●修改，指定保存Zookeeper中的数据的目录，目录需要单独创建
dataLogDir=/usr/local/zookeeper-3.5.7/logs   ●添加，指定存放日志的目录，目录需要单独创建
clientPort=2181   #客户端连接端口
#添加集群信息
server.1=192.168.73.30:3188:3288
server.2=192.168.73.40:3188:3288
server.3=192.168.73.140:3188:3288

启动zookeeper

三台服务器同样操作  按照 30 40 140 的启动顺序开始启动
cd /usr/local/zookeeper/bin
./zkServer.sh start
netstat -lntp | grep 2181

查看各服务器zookeeper状态

./zkServer.sh status

安装部署kafka集群

cd /opt/
tar zxvf kafka_2.13-2.8.2.tgz
mv kafka_2.13-2.8.2 /usr/local/kafka

修改配置文件

192.168.73.30
cd /usr/local/kafka/config/
cp server.properties{,.bak}
 
vim server.properties
broker.id=0    ●21行，broker的全局唯一编号，每个broker不能重复，因此要在其他机器上配置 broker.id=1、broker.id=2
listeners=PLAINTEXT://192.168.9.210:9092    ●31行，指定监听的IP和端口，如果修改每个broker的IP需区分开来，也可保持默认配置不用修改
num.network.threads=3    #42行，broker 处理网络请求的线程数量，一般情况下不需要去修改
num.io.threads=8         #45行，用来处理磁盘IO的线程数量，数值应该大于硬盘数
socket.send.buffer.bytes=102400       #48行，发送套接字的缓冲区大小
socket.receive.buffer.bytes=102400    #51行，接收套接字的缓冲区大小
socket.request.max.bytes=104857600    #54行，请求套接字的缓冲区大小
log.dirs=/usr/local/kafka/logs        #60行，kafka运行日志存放的路径，也是数据存放的路径
num.partitions=1    #65行，topic在当前broker上的默认分区个数，会被topic创建时的指定参数覆盖
num.recovery.threads.per.data.dir=1    #69行，用来恢复和清理data下数据的线程数量
log.retention.hours=168    #103行，segment文件（数据文件）保留的最长时间，单位为小时，默认为7天，超时将被删除
log.segment.bytes=1073741824    #110行，一个segment文件最大的大小，默认为 1G，超出将新建一个新的segment文件
zookeeper.connect=192.168.9.210:2181,192.168.9.120:2181,192.168.9.140:2181    ●123行，配置连接Zookeeper集群地址
 
 
 
 
#复制配置到其他两个服务器
cd /usr/local
scp -r kafka/ 192.168.73.40:/usr/local/
scp -r kafka/ 192.168.73.140:/usr/local/

修改环境变量

所有服务器操作
vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
suorce  /etc/profile

启动kafka

cd /usr/local/kafka/bin/
./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
netstat -lntp | grep 9092

 Kafka 命令行操作

kafka/bin/kafka-topics.sh --zookeeper IP1:2181,IP2:2181,IP3:2181 --create --topic 主题名 --partitions 分区数 --replication-factor 副本数      #创建topic
kafka/bin/kafka-topics.sh --zookeeper IP1:2181,IP2:2181,IP3:2181 --list                                             #查看topic列表
kafka/bin/kafka-topics.sh --zookeeper IP1:2181,IP2:2181,IP3:2181 --discribe --topic 主题名                          #查看topic详细信息
kafka/bin/kafka-topics.sh --zookeeper IP1:2181,IP2:2181,IP3:2181 --delete --topic 主题名                            #删除topic
kafka/bin/kafka-topics.sh --zookeeper IP1:2181,IP2:2181,IP3:2181 --alter --topic 主题名 --partitions 分区数         #修改topic的分区数（只能增不能减）
 
kafka/bin/kafka-console-producer.sh --broker-list IP1:9092,IP2:9092,IP3:9092 --topic 主题名                           #向topic推送数据
kafka/bin/kafka-console-consumer.sh --bootstrap-server IP1:9092,IP2:9092,IP3:9092 --topic 主题名 [--from-beginning]   #从topic拉取数据

搭建ELK+Filebead+zookeeper+kafka

1.部署 Filebeat 

cd /usr/local/filebeat
 
vim filebeat.yml
注释162、164行内容
163行起添加
output.kafka:
  enabled: true
  hosts: ["192.168.73.30:9092","192.168.73.40","192.168.73.140"]  #指定 Kafka 集群配置
  topic: "nginx"    #指定 Kafka 的 topic

浏览器访问filebeat，生成新日志数据

（http://192.168.73.130/test.html、http://192.168.73.130/test1.html、http://192.168.73.130/）

启动 filebeat
./filebeat -e -c filebeat.yml

2.Logstash配置

cd /etc/logstash/conf.d/
vim kafka.conf
input {
    kafka {
        bootstrap_server => "192.168.73.30:9092,192.168.73.40:9092,192.168.73.140:9092"
        topics => "nginx"
        type => "nginx_kafka"
        auto_offset_reset => "latest"
    }
}
 
#filter {}
 
output {
     elasticsearch {
          hosts => ["192.168.73.80:9200", "192.168.73.100:9200", "192.168.73.110:9200"]
          index => "nginx_kafka-%{+yyyy.MM.dd}"
     }
}
 
 
 
logstash -t -f kafka.conf
logstash -f kafka.conf

3.浏览器kibana访问验证