当前位置:   article > 正文

分布式消息队列--Kafka_kafka分布式消息队列

kafka分布式消息队列

定义与特性

Kafka是一个支持分区,多副本,且基于zookeeper协调的分布式消息系统。多个分区副本中只有Leader副本提供读写服务,其他Follower副本被动复制Leader的结果且不提供读写服务,目的主要是为了保证多副本数据与消费的一致性。

特性:用Scala语言编写的可以实时处理大数据相关需求场景,如基于hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎、访问日志、消息服务等。

在这里插入图片描述

使用场景

  1. 日志收集:收集各个服务的log,通过以统一接口服务的方式开放给各种消费者,如hadoop, Hbase, Solr等
  2. 消息系统:生产者消费者解耦、缓存消息等
  3. 运营指标:用来记录运营监控数据,包括各种分布式应用的数据,生产各种操作的几种反馈,如报警和报告
  4. 用户活动跟踪:用于记录web/app用户的各种活动

基本概念

名称 定义
Borker 消息中间件处理节点,一个Kafka节点就是一个Broker,一个或者多个Broker组成一个Kafka集群
Topic Kafka根据Topic对消息进行归类,发布到Kafka集群的每条消息都需要指定一个Topic
Partition 物理概念,一个Topic可以分为多个Partition,每个Partition内部消息是有序的
Producer 消息生产者,向Broker发送消息
Consumer 消息消费者,从Broker读取消息
ConsumerGroup 消费者组,每个消费者属于一个特定的消费者组,一条消息可以被多个不同的消费者组消费,但是一个消费者组中只能有一个消费者能够消息这条消息

基本使用

本地示例安装Kafka版本2.8.2,对应Scala版本2.12

环境准备
  1. 安装jdk
    yum install java-1.0.0-openjdk* -y

  2. 安装zookeeper
    wget https://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8.bin.tar.gz
    tar -zxvf apache-zookeeper-3.5.8-bin.tar.gz
    cd apache-zookeeper-3.5.8-bin
    cp conf/zoo_sample.cfg conf/zoo.cfg

  3. 启动zookeeper
    bin/zkServer.sh start conf/zoo.cfg
    bin/zkCli.sh -server 192.168.126.135:2181
    ls /

单机部署
  1. 下载安装包

cd /usr/local
#Scala版本2.12,Kafka版本2.8.2
wget https://mirrors.aliyun.com/apache/kafka/2.8.2/kafka_2.12-2.8.2.tgz
tar -zxf kafka_2.12-2.8.2.tgz
cd kafka_2.12-2.8.2/ && ll

在这里插入图片描述

  1. 修改conf/server.properties配置

vi config/server.properties

#broker.id属性在Kafka集群中必须唯一
broker.id=0
#Kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://192.168.126.135:9092
#Kafka消息存储文件路径
log.dirs=/usr/local/kafka_2.12-2.8.2/kafka-logs
#Kafka连接zookeeper地址
zookeeper.connect=192.168.126.135:2181

:wq

  1. 启动kafka
    kafka-server-start.sh [-daemon] server.properties
    (注意:启动kafka时会使用Linux主机名关联的IP地址,故需要把主机名和Linux的IP映射配置到本地host中,/etc/hosts)

启动命令:
bin/kafka-server-start.sh -daemon config/server.properties

bin/kafka-server-start.sh config/server.properties &
#查看zookeeper中kafka节点
bin/zkCli.sh
ls /brokers/ids

在这里插入图片描述

  1. 停止Kafka

bin/kafka-server-stop.sh

基本使用
  1. 主题操作

#创建一个分区数与备份因子均为1且名称为test的Topic (注意:若Producer发消息到某个指定的且不存在的Topic,则kafka会自动创建)
bin/kafka-topics.sh --create --zookeeper 192.168.126.135:2181 --topic test --partitions 1 --replication-factor 1

在这里插入图片描述

#查看kafka中存在的topic列表
bin/kafka-topics.sh --list --zookeeper 192.168.126.135:2181

在这里插入图片描述

#删除主题
bin/kafka-topics.sh --delete --zookeeper 192.168.126.135:2181 --topic test

在这里插入图片描述

  1. 发送与消费消息

#往指定kafka集群的指定topic发送消息
bin/kafka-console-producer.sh --broker-list 192.168.126.135:9092 --topic jeffrey

在这里插入图片描述

通过Offset Explorer工具查看kafka消息如下:
在这里插入图片描述

#从指定kafka集群的指定topic消费最新消息(默认)
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --topic jeffrey

在这里插入图片描述

#从指定kafka集群的指定topic从头开始消费消息
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --topic jeffrey --from-beginning

在这里插入图片描述

#从指定kafka集群的指定多个topic消费消息[从头开始消费]
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --whitelist “jeffrey|qiutee” --from-beginning
在这里插入图片描述

  1. 单播消息与多播消息

#单播消息(一个topic中的一个消息只能被同一个消费者组下的某一个消费者消费类似Queue模式,实现只需将所有消费者置于同一个消费者组即可)
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --topic jeffrey --consumer-property group.id=zfy-consumer-group

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

#多播消息(一个消息能被多个消费者消费类似发布订阅模式,实现只需将所有消费者置于不同消费者组即可)
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --topic jeffrey --consumer-property group.id=wqt-consumer-group

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

  1. 消费者组操作

#查看指定kafka集群中消费者组列表
bin/kafka-consumer-groups.sh --list --bootstrap-server 192.168.126.135:9092

在这里插入图片描述

#查看指定kafka集群的指定消费者组的描述信息(可查询消息消费情况)
bin/kafka-consumer-groups.sh --describe --bootstrap-server 192.168.126.135:9092 --group zfy-consumer-group
(topic:当前主题;partition:当前分区;current-offset:当前消费者组在当前主题当前分区中的已消费偏移量;log-end-offset:当前主题对应当前分区中消息的结束偏移量HW;lag:当前消费者组未消费的消费数,值为当前主题分区的LOG-END-OFFSET(LEO)减去当前消费者组已消费偏移量CURRENT-OFFSET)

在这里插入图片描述

主题与分区

Topic是消息分类的名称,一个Topic下可以有多个Partition日志文件。
Partition是一个有序消息序列,一个Partition对应一个Commit log文件,Partition中的offset唯一标识某个分区中的消息;一个Partition中的offset是唯一的,不同Partition的offset可能相同。
Kafka消息日志默认保留时间168h(近一周),可通过配置log.retention.hours自定义保留时间。
每个consumer是基于自己在commit log中的offset来进行消费的,消费offset由consumer自身维护;一般而言partition中的消息是按照顺序消费的,也可以通过offset重复消费或跳过消费。

#创建多个分区的主题
bin/kafka-topics.sh --create --zookeeper 192.168.126.135:2181 --topic jack --partitions 2 --replication-factor 1

在这里插入图片描述

#查看topic描述信息
bin/kafka-topics.sh --describe --zookeeper 192.168.126.135:2181 --topic jack

topic描述信息介绍:

  • leader节点负责给定partition的所有读写请求
  • replicas 表示某个partition在哪几个broker上存在备份。不管这几个是不是”leader“,甚至这个节点挂了也会列出
  • isr 是replicas的一个子集,它只列出当前还存活着的,并且已同步备份了该partition的节点

在这里插入图片描述

#增加topic的分区数量(注意:topic的分区数量只能增加不能减少,减少topic分区会报错)
bin/kafka-topics.sh --alter --zookeeper 192.168.126.135:2181 --topic jack --partitions 3

在这里插入图片描述

为什么要对Topic下数据进行分区存储?

1、数据分布式存储,commit log文件会受到所在机器的文件系统大小的限制,分区之后可以将不同的分区放在不同的机器上,理论上一个topic可以处理任意数量的数据。
2、为了提高并行度。

集群部署

企业级Kafka集群一般是多个broker实例部署到多台不同的机器节点上,本次演示由于资源有限,将在同一台虚拟机上部署三个broker实例组成一个Kafka伪集群。

集群搭建步骤如下:

1. 环境依赖

依据上述 基本使用-环境准备 章节介绍,确保集群部署前安装完成jdk,安装与启动完成zookeeper

2. 下载安装Kafka

依据上述 单机部署-下载安装包章节介绍,下载解压Kafka安装包并进入解压后的目录

3. 创建kafka集群配置目录与日志存储目录

#创建kafka集群配置目录
mkdir -p /usr/local/kafka-cluster/node1/config
mkdir -p /usr/local/kafka-cluster/node2/config
mkdir -p /usr/local/kafka-cluster/node3/config
#创建kafka集群日志存储目录
mkdir -p /usr/local/kafka-cluster/node1/kafka-logs
mkdir -p /usr/local/kafka-cluster/node2/kafka-logs
mkdir -p /usr/local/kafka-cluster/node3/kafka-logs

3. 拷贝并修改配置文件

拷贝三份config/server.properties配置文件到kafka集群配置目录下并进行相应的配置修改

cp /usr/local/kafka_2.12-2.8.2/config/server.properties /usr/local/kafka-cluster/node1/config/
cp /usr/local/kafka_2.12-2.8.2/config/server.properties /usr/local/kafka-cluster/node2/config/
cp /usr/local/kafka_2.12-2.8.2/config/server.properties /usr/local/kafka-cluster/node3/config/

分别对node1~node3下的config/server.properties的配置文件进行相应修改(下面以node1节点配置为例)

vi /usr/local/kafka-cluster/node1/config/server.properties

#集群broker id(注意:确保集群中每个broker实例都是唯一的)
broker.id=1
#当前broker实例部署的机器IP和服务提供的端口
listeners=PLAINTEXT://192.168.126.135:9093
#当前broker实例节点的日志存储目录(包含消息日志、索引日志以及时间索引日志)
log.dirs=/usr/local/kafka-cluster/node1/kafka-logs
#zk连接地址(注意:若存在zk集群,则多个zk节点地址用逗号隔开即可)
zookeeper.connect=192.168.126.135:2181

:wq

4. 启动kafka集群

分别启动三个broker实例即可

bin/kafka-server-start.sh -daemon /usr/local/kafka-cluster/node1/config/server.properties
bin/kafka-server-start.sh -daemon /usr/local/kafka-cluster/node2/config/server.properties
bin/kafka-server-start.sh -daemon /usr/local/kafka-cluster/node3/config/server.properties

5. 验证集群

#查看zk中三个broker实例是否都已经注册完成
/usr/local/zookeeper-3.5.10/bin/zkCli.sh -server 192.168.126.135:2181
ls /brokers/ids

在这里插入图片描述

quit

6. 集群操作演示

#创建一个分区数为2备份因子为3的名称为test-cluster-topic的主题(备份因子为3表示该topic的单个分区的总副本数为3,包含分区一个leader副本和两个follower副本)
bin/kafka-topics.sh --create --zookeeper 192.168.126.135:2181 --topic test-cluster-topic --partitions 2 --replication-factor 3

#查看新创建的topic的描述信息
bin/kafka-topics.sh --describe --zookeeper 192.168.126.135:2181 --topic test-cluster-topic
在这里插入图片描述
注:图中第一行描述当前topic相关信息;下面每行表示该topic每个分区的信息,其中Leader表示当前分区的leader副本所在的broker节点,Replicas表示当前分区在集群中哪些broker节点存在备份(包含未存活的broker节点以及非leader副本所在的broker节点),Isr则表示当前集群中存活着的并且已同步备份了当前分区的broker节点;Leader副本所在broker节点负责处理消息的读写请求,而Follower副本对应broker则是接收Leader副本的broker发送过来的消息并对消息进行备份存储。

#发送消息
bin/kafka-console-producer.sh --broker-list 192.168.126.135:9093,192.168.126.135:9094,192.168.126.135:9095 --topic test-cluster-topic
#消费消费
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9093,192.168.126.135:9094,192.168.126.135:9095 --topic test-cluster-topic --from-beginning

分区容错性测试:

由上图可知,主题test-cluster-topic分区0的Leader副本对应broker-1实例节点,为了演示分区容错,这里将分区0的Leader副本对应的broker节点进程杀掉,看是否会选举出新的broker节点作为分区0的Leader副本所对应的broker节点
ps -ef|grep kafka
kill 119259(这里我的broker-1实例进程id为119259)
等待broker-1进程杀掉后再次查看主题test-cluster-topic的描述信息
bin/kafka-topics.sh --describe --zookeeper 192.168.126.135:2181 --topic test-cluster-topic
在这里插入图片描述
可以看到,根据分区0 Isr中已存活且已同步备份的节点集来看,新选举出的主题test-cluster-topic分区0的Leader副本对应的实例节点为broker-3
(说明:一般情况同一主题下不同分区的Leader副本对应节点是不一样的,但在选举过程中,偶尔可能会出现同一主题下不同分区的Leader副本对应相同的broker节点)

查看zk中topic分区的state
get /brokers/topics/test-cluster-topic/partitions/0/state
在这里插入图片描述
(说明:Kafka相关集群信息保存在zk中,保持自身无状态,有利于水平扩容)

代码实战

producer
public class KfkProducer {
   
    private static final String CONNECT_STR = "192.168.126.129:9092";
    private static final String KFK_TOPIC = "test";
    private static final CountDownLatch CDL = new CountDownLatch(1);
    public static void main(String[
  • 1
  • 2
  • 3
  • 4
  • 5
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/很楠不爱3/article/detail/415659
推荐阅读
相关标签
  

闽ICP备14008679号