什么是消息队列
消息队列（Message Queue）是一种通信方法，用于在分布式系统中异步地传递消息。消息队列通过消息中间件将消息从发送者（Producer）传递给接收者（Consumer），允许应用程序解耦并提高系统的可扩展性和可靠性。
消息队列的特征
- 异步通信：消息队列允许发送者和接收者独立操作，发送者可以在发送消息后立即返回，不必等待接收者处理。
- 解耦：通过消息队列，发送者和接收者不必直接通信，从而减少了系统之间的依赖。
- 可靠性：消息队列系统通常提供消息持久化和重试机制，确保消息不会丢失。
- 可扩展性：通过消息队列，系统可以容易地扩展，以处理大量的并发消息。
- 负载均衡：消息队列可以将消息分发给多个消费者，从而实现负载均衡。
- 顺序保证：某些消息队列系统可以保证消息的顺序。
为什么需要消息队列
- 异步处理：允许后端系统异步处理任务，提高系统响应速度。
- 流量削峰：通过消息队列可以缓冲高峰流量，防止系统过载。
- 解耦系统：使得系统组件可以独立开发、部署和维护，提高系统的灵活性。
- 提高可靠性：通过持久化消息，确保在系统故障时消息不会丢失。
Kafka 基本概念
Kafka 是一种分布式消息队列系统，旨在处理高吞吐量的实时数据流。它由 LinkedIn 开发，并已捐赠给 Apache 基金会，成为开源项目。
Kafka 角色术语
- Producer（生产者）：发布消息到一个或多个 Kafka 主题。
- Consumer（消费者）：订阅一个或多个主题并处理发布到这些主题的消息。
- Broker：Kafka 集群中的一台服务器，负责接收和存储消息。
- Topic：消息的分类，每个主题可以有多个分区。
- Partition：主题的子单位，每个分区是一个有序的、不可变的消息序列。
- Replica：分区的副本，用于容错和高可用性。
- Leader：每个分区都有一个 Leader，负责所有读写请求。
- 追随者：从 Leader 副本同步数据的副本。
Topic 和 Partition
- Topic：Kafka 中消息的类别，可以理解为一个消息队列的名称。
- Partition：Topic 的分区，一个 Topic 可以有多个分区，分区是 Kafka 实现高吞吐量和并行处理的关键。
Producer 生产机制
- 消息生产:P roducer 将消息发布到一个或多个 Topic。
- 分区选择:P roducer 可以根据键值或使用默认分区器将消息分配到特定的分区。
- 确认机制:P roducer 可以配置不同的确认机制（例如 leader ack，all replicas ack）来确保消息的可靠传递。
Consumer 消费机制
- 订阅 Topic：Consumer 订阅一个或多个 Topic，Kafka 将消息推送给 Consumer。
- 消费位移：每个 Consumer 维护一个位移（offset），表示当前消费到的消息位置。
- 消费组：多个 Consumer 可以组成一个消费组（Consumer Group），Kafka 将消息分配给组内的 Consumer，实现负载均衡。
Zookeeper 概念介绍
ZooKeeper 是一个分布式协调服务，主要用于分布式应用程序中的配置管理、命名服务、分布式同步和组服务。
Zookeeper 的工作原理
- 数据模型：ZooKeeper 使用树状的数据模型，数据存储在 znode 中，类似于文件系统。
- 一致性保证：ZooKeeper 保证顺序一致性、原子性、单一系统映像和可靠性。
- Leader 选举：ZooKeeper 集群通过 Leader 选举保证数据的一致性和容错性。Leader 负责处理所有写请求，Follower 处理读请求。
- 会话管理：ZooKeeper 通过会话管理保持客户端连接的状态，检测并处理客户端故障。
- 通知机制：客户端可以对特定的 znode 设置监视器，当 znode 状态变化时，ZooKeeper 会通知相应的客户端。
单节点部署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
扩展：将远程的logstash输出到kafka中
[root@logstash1 logstash]# vim conf.d/beats.conf
input {
beats {
port => "5044"
codec => "json"
}
}
output {
kafka {
bootstrap_servers => "192.168.10.114:9092,192.168.10.115:9092,192.168.10.116:9092"
topic_id => "httpd-filebeat"
batch_size => "5"
codec => "json"
}
}
在logstash中运行配置
[root@logstash1 logstash]# logstash -f /etc/logstash/conf.d/beats.conf
生产消息（任意一个节点）
bin/kafka-console-producer.sh --broker-list kafka1:9092 -topic test
消费消息（任意一个节点）
bin/kafka-console-consumer.sh --bootstrap-server kafka2:9092 --topic test
删除topic（任意一个节点）
bin/kafka-topics.sh --delete --zookeeper kafka1:2181 --topic test
重启kafka
/etc/kafka/bin/kafka-server-stop.sh
/etc/zookeeper/bin/zkServer.sh restart
/etc/kafka/bin/kafka-server-start.sh config/server.properties &
问题解决
错误提示：
Failed to acquire lock on file .lock in /tmp/kafka-logs. A Kafka instance in another process or thread is using this directory.
解决方法：
[root@kafka1 kafka]# rm -rf /tmp/kafka-logs/*
[root@kafka1 kafka]# netstat -lnp|grep 9092
pkill杀死kafka的进程号
[root@kafka2 kafka]# ./bin/kafka-server-start.sh config/server.properties &
