zookeeper+kafka群集

什么是消息队列

消息队列（Message Queue）是一种通信方法，用于在分布式系统中异步地传递消息。消息队列通过消息中间件将消息从发送者（Producer）传递给接收者（Consumer），允许应用程序解耦并提高系统的可扩展性和可靠性。

消息队列的特征

1. 异步通信：消息队列允许发送者和接收者独立操作，发送者可以在发送消息后立即返回，不必等待接收者处理。
2. 解耦：通过消息队列，发送者和接收者不必直接通信，从而减少了系统之间的依赖。
3. 可靠性：消息队列系统通常提供消息持久化和重试机制，确保消息不会丢失。
4. 可扩展性：通过消息队列，系统可以容易地扩展，以处理大量的并发消息。
5. 负载均衡：消息队列可以将消息分发给多个消费者，从而实现负载均衡。
6. 顺序保证：某些消息队列系统可以保证消息的顺序。

为什么需要消息队列

1. 异步处理：允许后端系统异步处理任务，提高系统响应速度。
2. 流量削峰：通过消息队列可以缓冲高峰流量，防止系统过载。
3. 解耦系统：使得系统组件可以独立开发、部署和维护，提高系统的灵活性。
4. 提高可靠性：通过持久化消息，确保在系统故障时消息不会丢失。

Kafka 基本概念

Kafka 是一种分布式消息队列系统，旨在处理高吞吐量的实时数据流。它由 LinkedIn 开发，并已捐赠给 Apache 基金会，成为开源项目。

Kafka 角色术语

1. Producer（生产者）：发布消息到一个或多个 Kafka 主题。
2. Consumer（消费者）：订阅一个或多个主题并处理发布到这些主题的消息。
3. Broker：Kafka 集群中的一台服务器，负责接收和存储消息。
4. Topic：消息的分类，每个主题可以有多个分区。
5. Partition：主题的子单位，每个分区是一个有序的、不可变的消息序列。
6. Replica：分区的副本，用于容错和高可用性。
7. Leader：每个分区都有一个 Leader，负责所有读写请求。
8. 追随者：从 Leader 副本同步数据的副本。

Topic 和 Partition

• Topic：Kafka 中消息的类别，可以理解为一个消息队列的名称。
• Partition：Topic 的分区，一个 Topic 可以有多个分区，分区是 Kafka 实现高吞吐量和并行处理的关键。

Producer 生产机制

1. 消息生产:P roducer 将消息发布到一个或多个 Topic。
2. 分区选择:P roducer 可以根据键值或使用默认分区器将消息分配到特定的分区。
3. 确认机制:P roducer 可以配置不同的确认机制（例如 leader ack，all replicas ack）来确保消息的可靠传递。

Consumer 消费机制

1. 订阅 Topic：Consumer 订阅一个或多个 Topic，Kafka 将消息推送给 Consumer。
2. 消费位移：每个 Consumer 维护一个位移（offset），表示当前消费到的消息位置。
3. 消费组：多个 Consumer 可以组成一个消费组（Consumer Group），Kafka 将消息分配给组内的 Consumer，实现负载均衡。

Zookeeper 概念介绍

ZooKeeper 是一个分布式协调服务，主要用于分布式应用程序中的配置管理、命名服务、分布式同步和组服务。

Zookeeper 的工作原理

1. 数据模型：ZooKeeper 使用树状的数据模型，数据存储在 znode 中，类似于文件系统。
2. 一致性保证：ZooKeeper 保证顺序一致性、原子性、单一系统映像和可靠性。
3. Leader 选举：ZooKeeper 集群通过 Leader 选举保证数据的一致性和容错性。Leader 负责处理所有写请求，Follower 处理读请求。
4. 会话管理：ZooKeeper 通过会话管理保持客户端连接的状态，检测并处理客户端故障。
5. 通知机制：客户端可以对特定的 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 &