zookeeper+kafka
一、zookeeper
1.概述
zoo: 开源的分布式框架协调服务
zookeeper的工作机制：基于观察者模式设计的分布式结构，负责存储和管理架构当中的元信息，架构当中的应用接受观察者的监控，一旦数据有变化，通知对应的zookeeper，保存变化的信息。
2.特点
- 最少要有三台机器，一个领导者（leader)，多个跟随者（follower）
- zookeeper要有半数以上的节点存活，整个架构就可以正常工作，所以都是奇数台部署
- 全局数据一致
- 数据更新的原子性：要么都成功，要么都失败
- 实时性
3.zookeeper的数据架构
- 统一命名服务：不是以ip来记录，可以用域名也可以是主机名来记录信息。
- 统一配置管理，所有的节点信息的配置要是一致。
- 统一集群管理，在整个分布式的环境中，必须实时的掌握每个节点的状态，如果状态发生变化，要及时更新。
4. 实验
192.168.206.30
192.168.206.50
192.168.206.55
tar -xf apache-zookeeper-3.5.7-bin.tar.gz
apt -y install openjdk-8-jre-headless #安装依赖环境
java -version
mv apache-zookeeper-3.5.7-bin zookeeper
cd /opt/zookeeper/conf
cp zoo_sample.cfg zoo.cfg
vim zoo.cfg
#如下图所示
server.1=192.168.233.30:3188:3288
server.2=192.168.233.50:3188:3288
server.3=192.168.233.55:3188:3288
1:指的是服务器的序号
192.168.233.30 对应序号的服务器的ip地址
3188 leader和follower之间交换信息的端口
3288 选举端口
mkdir /opt/zookeeper/data
mkdir /opt/zookeeper/logs
useradd -m -s /sbin/nologin zookeeper
cd /opt
chown -R zookeeper.zookeeper zookeeper/
cd /opt/zookeeper/data
#三台机器的myid不同
echo 1 > myid
echo 2 > myid
echo 3 > myid
cd /opt/zookeeper/bin
./zkServer.sh start
./zkServer.sh status
二、kafka消息队列
1.概述
服务端向客户端发送一个指令，客户端收到指令，并且通过这个指令反馈到服务端，完成一个异步方式的通信的流程
- kafka消息队列---->大数据场景非常合适
- rabbitMQ--->适合处理小场景的数据请求
- activeMQ--->适合处理小场景的数据请求
2. 消息队列的应用场景
（1）异步处理*
用户的短信验证码，邮件通知
（2）系统解耦
微服务架构中的服务之间通信
降低各个组件之间的依赖程度（耦合度），提高组件的灵活性和可维护性
（3）负载均衡
高并发系统中的任务处理
消息队列把多个任务分发到多个消费者实例，电商平台的订单系统
（4）流量控制和限流
根据API请求来处理，通过控制消息的生产速度和消费的处理来完成限流。
端口：应用和应用之间通信
API接口：应用程序内部各个组件通信的方式（代码）
（5）任务调度和定时任务
消息队列可以定时的进行任务调度，按照消费者的计划生成对应的数据
（6）数据同步和分发*
日志收集和数据收集
（7）实时数据处理
（8）备份和恢复
3.消息队列的模式
（1）点对点
即一对一，生产者生产消息，消费者消费消息。
（2）发布/订阅模式
消息的生产者发布一个主题（topic），其他的消费者订阅这个主题，从而实现一对多。
4.kafka组件的名称
（1）主题
topic 主题是kafka数据流的一个基本单元，类似于数据的管道，生产者将消息发布到主题，其他的消费者订阅主题，来消费消息，主题可以被分区，分区有偏移量
（2）生产者
将消息写入主题和分区
（3）消费者
从分区和主题当中接收发布的消息。一个消费者可以订阅多个主题
（4）分区
一个主题可以被分为多个分区，每个分区都是数据的有序子集。分区越多，消费者消费的速度越快，可以避免生产者的消费堆积。分区当中有偏移量，按照偏移量进行有序存储，消费者可以独立的读取每个分区的数据
如何读取分区的数据，一般是不考虑的。只有消息出现堆积的时候，会考虑扩展分区数。
（5）偏移量
消息在分区当中的唯一标识，根据偏移量指定消费者获取消息的位置
test1 test2 test3 分区1
0 1 2
（6）经纪人
zookeeper，存储分区的信息，kafka集群的元数据
5.实验
kafka在zookeeper的基础上才能搭建
192.168.206.60 es1
192.168.206.70 es2
192.168.206.80 logstash kibana
192.168.206.40 filebeat
192.168.206.30 kafka1
192.168.206.50 kafka2
192.168.206.55 kafka3
-------------------------------------------------------------------------------------------------------------------------
cd /usr/local/filebeat
vim filebeat.yml
把tags和service_name改一下
把output.logstash两行注释掉
新增：
output.kafka:
enabled: true
hosts: ["192.168.206.30:9092","192.168.206.50:9092","192.168.206.55:9092"]
topic: nginx_mysql
ps -elf | grep filebeat
./filebeat -e -c filebeat.yml
cd /etc/logstash/conf.d
vim kafka.conf
input {
kafka {
bootstrap_servers => "192.168.206.30:9092,192.168.206.50:9092,192.168.206.55:9092"
topics => "nginx_mysql"
type => "nginx_kakfa"
codec => "json"
#指定数据的格式是json
auto_offser_reset => "latest"
#latest，从尾部开始 earliest 从头开始拉取
decorate_events => true
#传递给es的数据额外的附加kafka的属性数据
}
}output {
if "nginx-1" in [tags] {
elasticsearch {
hosts => ["192.168.206.60:9200","192.168.206.70:9200"]
index => "%{[fields][service_name]}-%{+YYYY.MM.dd}"}
}if "mysql-1" in [tags] {
elasticsearch {
hosts => ["192.168.206.60:9200","192.168.206.70:9200"]
index => "%{[fields][service_name]}-%{+YYYY.MM.dd}"}
}}
logstash -f kafka.conf --path.data /opt/testa &
kafka-console-consumer.sh --bootstrap-server 192.168.206.30:9092,192.168.206.50:9092,192.168.206.55:9092 --topic nginx_mysql --from-beginning
#看有没有获取
