zookeeper+kafka的消息队列
-
zookeeper+Kafka
-
两个都是消息队列的工具
-
消息队列
-
出现原因：生产者产生的消息与消费者处理消息的效率相差很大。为了避免出现数据丢失而设立的中间件。
-
在消息的生产者与消费之间设置一个系统，负责缓存生产者与消费者之间的消息的缓存。将消息排序。
-
优点：
-
解耦
-
将生产者与消费者的关系解开，这样避免一端的程序出问题导致整个两端程序都不可用。
-
-
冗余
-
为生产者与消费者之间产生的数据做了冗余
-
-
扩展性
-
利用zookeeper可以具备良好的扩展性。
-
-
灵活性
-
消息队列的对象可以是各种不同的程序
-
-
峰值处理能力
-
生产者产生的消息如果具有波动性，某一时刻峰值很高，利用消息队列就可以承载压力，然后慢慢分配给消息的消费者。
-
-
可恢复性
-
可以采用群集备份
-
-
顺序保证
-
消息以排序的方式存在
-
-
缓存
-
在生产者与消费者之间起到了一个缓冲的作用
-
-
异步通信
-
生产者产生的消息与消费者消费消息可以不再同一时间。
-
-
-
-
消息队列产品
-
kafka
-
apache旗下的产品
-
一种高吞吐量分布式发布/订阅消息队列的开源系统
-
kafka术语
-
broker
-
每一个安装了kafka的服务器都可以称为一个broker
-
-
topic
-
一类消息的集合被称为主题，一个broker可以有多个topic。
-
-
producer
-
消息生产者
-
-
consumer
-
消息消费者
-
-
partition
-
存放消息的区域，每一个topic可以有多个partition，但是处理还是一个一个区域处理。每一个区域内的消息都会被分配一个id。offset （偏移量）
-
-
consumer GROUP
-
消费者组
-
-
messages
-
消息，通信的基本单位。
-
-
-
kafka架构
-
-
需要注意的是，kafka作为一个支持多生产者多消费者的架构，再写入消息时允许多个生产者写道同一个partition，但是消费者读取的时候一个partition仅允许一个消费者消费，但一个消费者可以消费多个partition。
-
kafka支持消息持久化存储，消息会被存储在缓冲区内，达成一定条件就会被写入磁盘。在写入磁盘是是按照顺序的。
-
topic与partition的关系
-
partition的数量决定了组成topic的log的数量， 因此推荐partition的数量要大于同时允许的consumer数量，要小于等于集群broker的数量。
-
-
-
zookeeper
-
一种分布式协调技术，用于解决分布式环境中多个进程之间的同步控制，让他们有序的访问某种共享资源，尽量减少脑裂的发生。
-
开源的分布式协调服务
-
-
解决问题
-
单点故障
-
-
工作原理
-
master启动
-
节点向zookeeper注册信息，在选举产生master时采用最小编号算法，既最先的编号最小。
-
-
master故障
-
出现故障的节点信息会被自动删除，再通过选举投票产生新的master。
-
zookeeper判断节点故障的依据： 节点必须与zookeeper保持心跳连接。 节点能够及时响应zookeeper的请求。
-
-
-
master恢复
-
master恢复后，会再次向zookeeper注册自身节点信息，编号就会变淡。
-
-
-
架构图
-
-
四个角色
-
leader：领导者
-
负责发起投票与决议，更新系统状态
-
3888，选举通信端口
-
-
-
follower：追随者
-
负责请求客户端的请求并将结果给客户端，在选举过程中参与投票。
-
-
observer：观察者角色
-
负责接收客户端请求，并将请求转发给leader。但不参与投票，目的是扩展系统，提高伸缩性。
-
watcher：观察者，监视者
-
sentinel：哨兵
-
-
-
client：客户端
-
负责向zookeeper发起请求。
-
-
-
-
zookeeper在kafka中的作用
-
broker注册
-
每一个kafka的master节点都会在zookeeper中注册信息，提供id，每个master整个id都必须唯一。
-
-
topic注册
-
每一个主题也会在zookeeper注册信息。
-
-
生产者与消费者的负载均衡（实际上是高可用）
-
分区与消费者的联系
-
消费进度offset记录
-
消费者注册
-
根据注册信息将消费者分配到不同的分区，以达到负载均衡的目的。
-
-
-
-
-
zookeeper+kafka部署
-
本案例以zookeeper3.6版本与kafka2.13版本为例
-
单节点部署kafka进行测试
-
可以设置hosts文件进行域名通信
-
安装依赖环境
-
yum -y install java 注意是1.8版本。
-
-
解压即可用移动到/etc/zookeeper
-
修改配置文件
-
可以单独指定数据目录
-
dataDir=/etc/zookeeper/zookeeper-data
-
然后将指定的目录创建出来
-
启动查看状态
-
./bin/zkServer.sh start
-
-
-
-
-
-
单节点部署zookeeper+kafka做测试
-
部署zookeeper
-
可以设置hosts文件进行域名通信
-
安装依赖环境
-
yum -y install java 注意是1.8版本。
-
-
解压即可用移动到/etc/zookeeper
-
修改配置文件
-
可以单独指定数据目录
-
dataDir=/etc/zookeeper/zookeeper-data
-
然后将指定的目录创建出来
-
之后启动查看状态
-
-
-
-
-
部署kafka
-
同样的解压即可用移动到/etc/kafka
-
同样的可以修改配置文件指定日志文件目录并将目录创建出来
-
log.dirs=/etc/kafka/kafka-logs
-
mkdir /etc/kafka/kafka-logs
-
-
启动kafka需要指定配置文件并可以设置为后台启动。
-
bin/kafka-server-start.sh config/server.properties &
-
-
-
需要注意启动时要先启动zookeeper，结束时要先结束kafka。
-
测试
-
创建topic
-
kafka-topics.sh --create --zookeeper kafka1:2181 --replication-factor 1 --partitions 1 --topic test
-
kafka-topic.sh --create --zookeeper broker域名或IP:端口一般2181 --replication-factor 集群master数量 --partitions 1 topic中分区数量 --topic 分区名
-
-
列出topic
-
kafka-topics.sh --list --zookeeper brokerIP/域名:端口(2181)
-
-
查看topic
-
kafka-topics.sh --describe --zookeeper kafka1:2181 --topic test
-
describ ：描述
-
-
-
生产消息进入命令行模式
-
/kafka-console-producer.sh --broker-list kafka1:9092 -topic tes
-
9092kafka监听客户端连接的端口
-
-
-
消费消息，进入查看产生消费消息的界面
-
kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test
-
-
删除topic
-
kafka-topic.sh --delete --zookeeper Kafka服务器：端口 --topic 主题名
-
-
-
-
集群部署zookeeper+kafka
-
修改主机hosts文件，使之可以使用域名访问
-
安装zookeeper三台主机安装部署略有不同
-
准备前奏
-
防火墙与内核机制处理
-
安装依赖环境java
-
-
解压移动
-
tar zxvf apache-zookeeper-3.6.0-bin.tar.gz
-
mv apache-zookeeper-3.6.0-bin /etc/zookeeper
-
-
创建数据保存目录
-
mkdir zookeeper-data
-
-
修改配置文件
-
dataDir=/etc/zookeeper/zookeeper-data
-
指定数据保存目录
-
-
clientPort=2181
-
确定监听端口
-
-
添加节点注册信息有几个添加几个
-
server.1=192.168.10.101:2888:3888
-
2888：集群内部通信端口，主要用于选举leader
-
3888：其他服务器在选举leader时监听的端口。
-
-
-
-
创建id文件
-
echo '1' > /etc/zookeeper/zookeeper-data/myid
-
每个节点IP不同
-
-
-
最后启动并查看状态
-
-
安装kafka，节点配置相同
-
解压并移动
-
tar zxvf kafka_2.13-2.4.1.tgz
-
mv kafka_2.13-2.4.1 /etc/kafka
-
-
修改配置文件
-
listeners=PLAINTEXT://192.168.10.101:9092
-
监听客户端连接
-
-
log.dirs=/etc/kafka/kafka-logs
-
指定日志文件位置
-
-
num.partitions=1
-
分区数量，不能超过节点数。
-
-
集群地址信息
-
zookeeper.connect=192.168.10.101:2181,192.168.10.102:2181,192.168.10.103:2181
-
-
-
创建日志目录
-
mkdir /etc/kafka/kafka-logs
-
-
最后启动查看状态
-
测试
-
在任意一节点上创建topic
-
列出任意一节点的topic
-
进行生产与消费的对应
-
-
ELK日志系统添加接口对接Kafka
-
input { beats { port => "5044" codec => "json" } } output { kafka { bootstrap_servers => "192.168.10.101:9092,192.168.10.102:9092,192.168.10.103:9092" topic_id => "httpd-filebeat" batch_size => "5" codec => "json" } }
-
-
-
该集群中用到的端口
-
zookeeper客户端连接的端口：2181
-
zookeerper内部通信端口：2888
-
zookeeper监听选举过程的端口：3888
-
kafka监听客户端的接口：9092
-
-
-
-
-
-
