zookeeper+kafka消息队列群集部署

一、消息队列

1：什么是消息队列

消息队列是一种在分布式系统中用于通信和数据传递的重要技术。
消息队列可以理解为一个按照特定规则存储消息的容器。当一个系统或应用程序产生消息（例如数据、指令、事件等）时，它将这些消息放入消息队列中。而其他需要处理这些消息的系统或应用程序，则从队列中取出消息进行处理。
消息队列的主要作用包括：

1.解耦：发送方和接收方不需要直接相互依赖，它们只与消息队列进行交互，降低了系统之间的耦合度。例如，在电商系统中，订单处理模块和库存管理模块可以通过消息队列进行通信，彼此不需要知道对方的具体实现细节。
2.异步处理：发送方可以快速将消息放入队列，无需等待接收方立即处理，从而提高系统的响应速度和性能。比如用户注册后发送欢迎邮件，注册流程无需等待邮件发送完成。
3.流量削峰：在高并发场景下，能够缓冲瞬时的大量请求，避免后端系统被压垮。比如电商促销活动时，大量订单请求可以先放入消息队列，后端系统按自身处理能力逐步处理。
4.保证消息的可靠传递：即使接收方暂时不可用，消息也会在队列中存储，待接收方恢复后进行处理。
常见的消息队列产品有 RabbitMQ、Kafka 等。

2.消息队列的特征

1. 异步处理：
   消息的发送者和接收者不需要同时在线或同步处理消息。例如，在电商系统中，用户下单后，订单处理的消息可以先放入消息队列，后续的库存更新、物流通知等操作可以异步进行，而不必等待每个环节依次完成，从而提高系统的响应速度和整体性能。

2. 解耦：
   发送方和接收方之间的依赖关系被大大降低。比如，一个系统中的支付模块和订单模块，如果直接相互调用，当其中一个模块出现问题或变更时，可能会影响到另一个模块。而通过消息队列，它们只需关注各自的业务逻辑，发送和接收相关消息即可，实现了模块之间的松耦合。

3. 削峰填谷：
   能够应对突发的流量高峰。以在线直播中的点赞功能为例，大量用户同时点赞时，产生的请求可以先放入消息队列，后端服务按照自身的处理能力逐步消费这些消息，避免系统因瞬时高并发而崩溃。

4. 可靠存储：
   确保消息在传输过程中不会丢失。即使接收方暂时不可用，消息也会在队列中存储，等待接收方恢复后进行处理。这在金融交易等对数据完整性要求高的场景中尤为重要。

5. 排序保证：
   按照特定的顺序处理消息。例如，在一些业务流程中，可能需要先处理创建订单的消息，再处理支付订单的消息，消息队列可以保证这种顺序性。

6. 广播功能：
   一条消息可以被多个消费者同时获取和处理。例如，系统中的监控模块和报警模块可能都需要获取系统故障的消息来执行各自的操作。
   综上所述，消息队列的这些特征使其在构建复杂的分布式系统中发挥着重要作用，能够有效地提高系统的可扩展性、可靠性和性能。

2：消息队列的特征

（1）存储

与依赖于使用套接字的基本 TCP 和 UDP 协议的传统请求和响应系统不同，消息队列通常将消息存储在某种类型的缓冲区中，直到目标进程读取这些消息或将其从消息队列中显式移除为止。

（2）异步

与请求和响应系统不同，消息队列通过缓冲消息可以在应用程序中公开一定程度的异步性，允许源进程发送消息并在队列中累积消息，而目标进程则可以挑选消息进行处理。 这样，应用程序就可以在某些故障情况下运行，例如连接断断续续或源进程或目标进程故障。

路由：消息队列还可以提供路由功能，其中多个进程可以在同一队列中读取或写入消息，从而实现广播或单播通信模式。

3：为什么需要消息队列

（1）解耦

允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

（2）冗余

消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

（3）扩展性

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。

（4）灵活性 & 峰值处理能力

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

（5）可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

（6）顺序保证：

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。（Kafka 保证一个 Partition 内的消息的有序性）

（7）缓冲

有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。

（8）异步通信

很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

1：****kafka基本概念

Kafka是一种高吞吐量的分布式发布/订阅消息系统，这是官方对kafka的定义，这样说起来，可能不太好理解，这里简单举个例子：现在是个大数据时代，各种商业、社交、搜索、浏览都会产生大量的数据。那么如何快速收集这些数据，如何实时的分析这些数据，是一个必须要解决的问题，同时，这也形成了一个业务需求模型，即生产者生产（produce）各种数据，消费者（consume）消费（分析、处理）这些数据。那么面对这些需求，如何高效、稳定的完成数据的生产和消费呢？这就需要在生产者与消费者之间，建立一个通信的桥梁，这个桥梁就是消息系统。从微观层面来说，这种业务需求也可理解为不同的系统之间如何传递消息。

kafka是Apache组织下的一个开源系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop平台的数据分析、低时延的实时系统、storm/spark流式处理引擎等。kafka现在它已被多家大型公司作为多种类型的数据管道和消息系统使用。

2：****kafka角色术语

kafka的一些核心概念和角色

1. Broker：Kafka集群包含一个或多个服务器，每个服务器被称为broker。
2. Topic：每条发布到Kafka集群的消息都有一个分类，这个类别被称为Topic（主题）。
3. Producer：指消息的生产者，负责发布消息到kafka broker。
4. Consumer：指消息的消费者，从kafka broker拉取数据，并消费这些已发布的消息。
5. Partition：Partition是物理上的概念，每个Topic包含一个或多个Partition，每个partition都是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。
6. Consumer Group：消费者组，可以给每个Consumer指定消费组，若不指定消费者组，则属于默认的group。
7. Message：消息，通信的基本单位，每个producer可以向一个topic发布一些消息。

群集部署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