您的位置:首页 > 科技 > IT业 > zookeeper+kafka群集

zookeeper+kafka群集

2024/10/4 1:06:29 来源:https://blog.csdn.net/yiluo__/article/details/140446372  浏览:    关键词: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 &

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com