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