zookeeper+kafka消息队列群集部署

什么是消息队列

▪ 消息（Message）是指在应用间传送的数据

▪ 消息队列（Message Queue）是一种应用间的通信方式解决方法，确保消息的可靠传递

消息队列的特征

▪ 存储

▪ 将消息存储在某种类型的缓冲区，知道目标进程读取这些消息或将其从消息队列中显式移除为止

▪ 异步

▪ 消息队列通过缓冲消息可以在应用程序中公开一定程度的异步性，允许源进程发送消息并在队列中累计消息，而目标进程则可以挑选消息进行处理

消息队列概念

▪ 为什么需要消息队列

▪ 解耦

▪ 冗余

▪ 扩展性

▪ 灵活性

▪ 峰值处理能力

▪ 可恢复性

▪ 顺序保证

▪ 异步通信

kafka基础概念

▪ 什么是kafka

▪ Kafka是一种高吞吐量的分布式发布/订阅消息系统

▪ Kafka是Apache组织下的一个开源系统

▪ 可以实时的处理大量数据以满足各种需求场景

Kafka角色术语

▪ Broker：Kafka集群包含一个或多个服务器，每个服务器被称为broker（经纪人）

▪ Topic：每条发布到Kafka集群的消息都有一个分类，这个类别被称为Topic （主题）

▪ Producer：指消息的生产者，负责发布消息到kafka broker

▪ Consumer：指消息的消费者，从kafka broker拉取数据，并消费这些已发布的消息

▪ Partition：Partition是物理上的概念，每个Topic包含一个或多个Partition，每个Partition都是一个有序的队列。Partition中的每条消息都会被分配一个有序的id（offset）

▪ Consumer Group：消费者组，可以给每个Consumer指定消费组，若不指定消费者组，则属于默认的Group

▪ Message：消息，通信的基本单位，每个producer可以向一个topic发布一些消息

Zookeeper基础概念

▪ Zookeeper是一种分布式协调技术，所谓分布协调技术主要是用来解决分布式环境中多个进程之间的同步控制，让他们有序的去访问某种共享资源，放置造成资源竞争（脑裂）的后果

Zookeeper工作原理

▪ master启动

▪ 各节点向Zookeeper中注册节点信息，以编号最小算法选举出一个主节点，另外的节点就是备用节点，由Zookeeper完成对两个master进程的调度，和主、备节点的分配和写作

▪ master故障

▪ 如果节点A发生了故障，这时候它在Zookeeper所注册的节点信息会被自动删除，并会再次发出选举

▪ master恢复

▪ 如果主节点恢复了，它会再次向Zookeeper注册自身的节点信息，但注册的节点信息编号会变小，因此不会再称为master，而是另一台节点继续担任master

zookeeper架构

Leader：领导角色，主要负责投票的发起和决议，以及更新系统状态

follower：跟随角色，用于接收客户端的请求并返回结果给客户端，在选举过程中参与投票

observer：观察者角色，用户接收客户端的请求，并将写请求转发给leader，同时同步leader状态，但是不参与投票。Obeserver目的是扩展系统，提高伸缩性

client：客户端角色，用于向zookeeper发起请求

zookeeper在kafka中的作用

▪ Broker注册

▪ Topic注册

▪ 生产者负载均衡

▪ 消费者负载均衡

▪ 记录消息分区与消费者的关系

▪ 消息消费进度Offset记录

▪ 消费者记录

实验

单节点部署kafka

主机：

kafka1:192.168.10.101

1：安装zookeeper

[root@kafka1 ~]# yum -y install java

[root@kafka1 ~]# tar zxvf apache-zookeeper-3.6.0-bin.tar.gz

[root@kafka1 ~]# mv apache-zookeeper-3.6.0-bin /etc/zookeeper

[root@kafka1 ~]# cd /etc/zookeeper/conf

[root@kafka1 ~]# mv zoo_sample.cfg zoo.cfg

[root@kafka1 ~]# vim zoo.cfg

dataDir=/etc/zookeeper/zookeeper-data

[root@kafka1 ~]# cd /etc/zookeeper/

[root@kafka1 kafka]# mkdir /etc/zookeeper/zookeeper-data/

[root@kafka1 zookeeper]# ./bin/zkServer.sh start

[root@kafka1 zookeeper]# ./bin/zkServer.sh status

2：安装kafka

[root@kafka1 ~]# tar zxvf kafka_2.13-2.4.1.tgz

[root@kafka1 ~]# mv kafka_2.13-2.4.1 /etc/kafka

[root@kafka1 ~]# cd /etc/kafka/

[root@kafka1 kafka]# vim config/server.properties

log.dirs=/etc/kafka/kafka-logs #60行

[root@kafka1 kafka]# mkdir /etc/kafka/kafka-logs

[root@kafka1 kafka]# bin/kafka-server-start.sh config/server.properties &

检查两个端口的开启状态

[root@kafka1 kafka]# netstat -anpt | grep 2181

[root@kafka1 kafka]# netstat -anpt | grep 9092

注意：启动时先启动zookeeper，关闭时先关闭kafka

如果要关闭zookeeper

[root@kafka1 zookeeper]# ./bin/zkServer.sh start

如果要关闭kafka

[root@192 kafka]# bin/kafka-server-stop.sh

如果关不了，就kill杀死该进程

3：测试

创建topic

bin/kafka-topics.sh --create --zookeeper kafka1:2181 --replication-factor 1 --partitions 1 --topic test

列出topic

bin/kafka-topics.sh --list --zookeeper kafka1:2181

查看topic

bin/kafka-topics.sh --describe --zookeeper kafka1:2181 --topic test

生产消息

bin/kafka-console-producer.sh --broker-list kafka1:9092 -topic test

消费消息

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test

删除topic

bin/kafka-topics.sh --delete --zookeeper kafka1:2181 --topic test

五：群集部署kafka

主机

kafka1：192.168.10.101

kafka2：192.168.10.102

kafka3：192.168.10.103

1：zookeeper的部署

（1）安装zookeeper（三个节点的配置相同）

[root@kafka1 ~]# yum -y install java

[root@kafka1 ~]# tar zxvf apache-zookeeper-3.6.0-bin.tar.gz

[root@kafka1 ~]# mv apache-zookeeper-3.6.0-bin /etc/zookeeper

（2）创建数据保存目录（三个节点的配置相同）

[root@kafka1 ~]# cd /etc/zookeeper/

[root@kafka1 zookeeper]# mkdir zookeeper-data

（3）修改配置文件（三个节点的配置相同）

[root@kafka1 zookeeper]# cd /etc/zookeeper/conf

[root@kafka1 ~]# mv zoo_sample.cfg zoo.cfg

[root@kafka1 ~]# vim zoo.cfg

dataDir=/etc/zookeeper/zookeeper-data

clientPort=2181

server.1=192.168.10.114:2888:3888

server.2=192.168.10.115:2888:3888

server.3=192.168.10.116:2888:3888

注释：zookeeper只用的端口

2181：对cline端提供服务

3888：选举leader使用

2888：集群内机器通讯使用（Leader监听此端口）

（4）创建节点id文件（按server编号设置这个id，三个机器不同）

节点1：

[root@kafka1 conf]# echo '1' > /etc/zookeeper/zookeeper-data/myid

节点2：

[root@kafka2 conf]# echo '2' > /etc/zookeeper/zookeeper-data/myid

节点3：

[root@kafka3 conf]# echo '3' > /etc/zookeeper/zookeeper-data/myid

（5）三个节点启动zookeeper进程

[root@kafka1 conf]# cd /etc/zookeeper/

[root@kafka1 zookeeper]# ./bin/zkServer.sh start

[root@kafka1 zookeeper]# ./bin/zkServer.sh status

2：kafka的部署

（1）kafka的安装（三个节点的配置相同）

[root@kafka1 ~]# tar zxvf kafka_2.13-2.4.1.tgz

[root@kafka1 ~]# mv kafka_2.13-2.4.1 /etc/kafka

（2）修改配置文件

[root@kafka1 ~]# cd /etc/kafka/

[root@kafka2 kafka]# vim config/server.properties

broker.id=1 ##21行 修改，注意其他两个的id分别是2和3

listeners=PLAINTEXT://192.168.10.114:9092 #31行 修改，其他节点改成各自的IP地址

log.dirs=/etc/kafka/kafka-logs ## 60行 修改

num.partitions=1 ##65行 分片数量，不能超过节点数

zookeeper.connect=192.168.10.114:2181,192.168.10.115:2181,192.168.10.116:2181

注释：

9092是kafka的监听端口

（3）创建日志目录（三个节点的配置相同）

[root@kafka1 kafka]# mkdir /etc/kafka/kafka-logs

（3）在所有kafka节点上执行开启命令，生成kafka群集（三个节点的配置相同）

[root@kafka1 kafka]# ./bin/kafka-server-start.sh config/server.properties &

如果启动不了，可以将/etc/kafka/kafka-logs中的数据清除再试试

3：测试

创建topic（任意一个节点）

bin/kafka-topics.sh --create --zookeeper kafka1:2181 --replication-factor 1 --partitions 1 --topic test

列出topic（任意一个节点）

bin/kafka-topics.sh --list --zookeeper kafka1:2181

bin/kafka-topics.sh --list --zookeeper kafka2:2181

bin/kafka-topics.sh --list --zookeeper kafka3:2181

生产消息

bin/kafka-console-producer.sh --broker-list kafka1:9092 -topic test

消费消息

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test

删除topic

bin/kafka-topics.sh --delete --zookeeper kafka1:2181 --topic test