Kafka学习笔记1(千峰教育)

2023-12-28 16:51:00

一、为什么使用消息队列

1.使用同步的通信方式来解决多个服务之间的通信

在这里插入图片描述

同步的通信方式会存在性能和稳定性的问题。

2.使用异步的通信方式

在这里插入图片描述

针对于同步的通信方式来说,异步的方式,可以让上游快速成功,极大提高了系统的吞吐量。而且在分布式系统中,通过下游多个服务的分布式事务的保障,也能保障业务执行之后的最终一致性。

消息队列解决具体的是什么问题——通信问题。

二、消息队列的流派

目前消息队列的中间件选型有很多种:

  • rabbitMQ:内部的可玩性(功能性)是非常强的
  • rocketMQ:阿里内部一个大神,根据kafka的内部执行原理,手写的一个消息队列中间件。性能是与kafka比肩,除此之外,在功能上封装了更多的功能。
  • kafka:全球消息处理性能最快的一款MQ
  • zeroMQ

这些消息队列中间件有什么区别?

1.有broker

  • 重topic:kafka、RocketMQ、ActiveMQ

    整个broker,依据topic来进行消息的中转。在重topic的消息队列里必然需要topic的存在。

  • 轻topic:RabbitMQ

    topic只是一种中转模式。

2.无broker

在生产者和消费者之间没有使用broker,例如zeroMQ,直接使用socket进行通信。

三、Kafka的基本知识

1.Kafk2a的安装

  • 部署一台zookeeper服务器

  • 安装jdk

  • 下载kafka的安装包:https://kafka.apache.org/downloads

  • 上传kafka到服务器上:/user/local/kafka

  • 解压缩压缩包

  • 进入到config目录内,修改server.properties

    #broker.id属性在kafka集群中必须是要唯一
    broker.id=0
    #kafka部署的机器ip和提供服务的端口号
    listeners=PLAINTEXT://服务器IP:9092
    #kafka的消息存储文件
    log.dir=/usr/local/data/kafka-logs
    #kafka连接zookeeper的地址
    zookeeper.connect=服务器IP:2181
    
  • 进入到bin目录内,执行以下命令来启动kafka服务器(带着配置文件)

    ./kafka-server-start.sh -daemon ../config/server.properties
    
  • 校验kafka是否启动成功:

    进入到zk内查看是否有kafka的节点:brokers/ids/0

2.Kafka中的一些基本概念

kafka中有这么些复杂的概念

在这里插入图片描述

名称解释
Broker消息中间件处理节点,一个Kafka节点就是一个broker,一个或多个Broker可以组成一个Kafka集群
TopicKafka根据topic对消息进行归类,发布到Kafka集群的每条消息都需要指定一个topic
Producer消息生产者:向Broker发送消息的客户端
Consumer消息消费者:从Broker读取消息的客户端

3.创建topic

  • 通过kafka命令向zk中创建一个主题

    ./kafka-topics.sh --create --zookeeper ip:2181 --replication-factor 1 --partitions 1 --topic test
    
  • 查看当前zk中所有的主题

    ./kafka-topics.sh --list --zookeeper ip:2181
    test
    

4.发送消息

把消息发送给broker中的某个topic,打开一个kafka发送消息的客户端,然后开始用客户端向kafka服务器发送消息

./kafka-console-producer.sh --broker-list ip:9092 --topic test

5.消费消息

打开一个消费消息的客户端,向kafka服务器的某个主题消费信息

  • 方式一:从当前主题中的最后一条消息的offset(偏移量位置)+1开始消费

    ./kafka-console-consumer.sh --bootstrap-server ip:9092 --topic test
    
  • 方式二:从当前主题中的第一条消息开始消费

    ./kafka-console-consumer.sh --bootstrap-server ip:9092 --from-beginning --topic test
    

6.关于消息的细节

在这里插入图片描述

  • 生产者将消息发送给broker,broker会将消息保存在本地的日志文件中

    /usr/local/kafka/data/kafka-logs/主题-分区/000000000.log
    
  • 消息的保存是有效的,通过offset偏移量来描述消息的有序性

  • 消费者消费信息时也是通过offset来描述当前要消费的那条消息的位置

7.单播消息

在一个kafka的topic中,启动两个消费者,一个生产者。问:生产者发送消息,这条消息是否同时会被两个消费者消费?

如果多个消费者在同一个消费组,那么只有一个消费者可以收到订阅的topic中的消息。换言之,同一个消费组中只能有一个消费者收到一个topic中的消息。

./kafka-console-consumer.sh --bootstrap-server ip:9092 --consumer-property gropu.id=testGroup --topic test

8.多播消息

不同的消费组订阅同一个topic,那么不同的消费组中只有一个消费者能收到消息。实际上也是多个消费组中的多个消费者收到了同一个消息。

#消费组testGroup1
./kafka-console-consumer.sh --bootstrap-server ip:9092 --consumer-property gropu.id=testGroup1 --topic test
#消费组testGroup2
./kafka-console-consumer.sh --bootstrap-server ip:9092 --consumer-property gropu.id=testGroup2 --topic test

在这里插入图片描述

9.查看消费组的详细信息

通过以下命令可以查看到消费组的详细信息:

./kafka-consumer-groups.sh --bootstrap-server ip:9092 --describe --group testGroup

重点关注以下几个信息:

  • Currennt-offset:当前消费组的已消费偏移量
  • Log-end-offset:主题对应分区消息的结束偏移量(HW)
  • Lag:当前消费组未消费的消息数

四、Kafka中主题和分区的概念

1.主题Topic

主题-topic在kafka中是一个逻辑的概念,kafka通过topic将消息进行分类。不同的topic会被订阅该topic的消费者消费。

但是有一个问题,如果说这个topic中的消息非常非常多,多到需要几T来存,因为消息是会保存到log日志文件中的。为了解决这个文件过大的问题,kafka提出了Partition分区的概念。

2.分区Partition

1)分区的概念

通过partition将一个topic中的消息来存储。这样的好处有多个:

  • 分区存储,可以解决统一存储文件过大的问题。

  • 提供了读写的吞吐量,读和写可以同时在多个分区中进行。

    在这里插入图片描述

2)创建多分区的主题

./kafka-topics.sh --create --zookeeper ip:2181 --replication-factor 1 --partitions 2 --topic test1

3.kafka中消息日志文件中保存的内容

  • 00000.log:这个文件中保存的就是消息。

  • _consumer_offsets-49

    kafka内部自己创建了_consumer_offset主题包含了50个分区。这个主题用来存放消费者消费某个主题的偏移量,因为每个消费者都会自己维护消费的主题的偏移量,也就是说每个消费者会把消费的主题的偏移量自主上报给kafka中的默认主题:consumer_offsets。因为kafka为了提升这个主题的并发性,默认设置了50个分区。

    • 提交到哪个分区:通过hash函数确定:hash(consumerGroupId)%_consumer_offsets主题的分区数。
    • 提交到该主题中的内容是:key是consumerGroupId+topic+分区号,value就说当前offset的值。
  • 文件中保存的消息,默认保存7天,7天到后消息会被删除。

五、Kafka集群操作

1.搭建kafka集群(三个broker)

  • 创建三个server.properties文件

    # 0 1 2
    broker.id=2
    # 9092 9093 9094
    listeners=PLAINTEXT://192.168.65.60:9004
    # kafka-logs kafka-logs-1 kafka-logs-2
    log.dir=/usr/local/data/kafka-logs-2
    
  • 通过命令来启动三台broker

    ./kafka-server-start.sh -daemon ../config/server.properties
    ./kafka-server-start.sh -daemon ../config/server1.properties
    ./kafka-server-start.sh -daemon ../config/server2.properties
    
  • 校验是否启动成功

    进入到zk中查看/brokers/ids中是否有3个znode(0,1,2)

2.副本的概念

在创建主题时,除了指明主题的分区数外,还指明了副本数,那么副本是一个什么概念呢?

副本是为了给主题中的分区创建多个备份,多个副本在kafka集群的多个broker中,会有一个副本作为leader,其他是follower。

在这里插入图片描述

  • leader:kafka的写和读的操作都发生在leader上。leader负责把数据同步给follower。当leader挂了,经过主从 选举,从多个follower中选举产生一个新的leader。
  • follower:接受leader同步的数据。
  • isr:可以同步和已同步的节点会被存入到isr集合中。细节:如果isr中的节点性能比较差,会被踢出isr集合。

(重点~!)此时,broker、主题、分区、副本这些概念就全部展现了,大家需要把这些概念梳理清除:

集群中有多个broker,创建主题时可以指明主题有多个分区(把消息拆分到不同的分区中存储),可以为分区创建多个副本,不同的副本存放在不同的broker里。

3.关于集群消费

1)向集群发送消息

./kafka-console-producer.sh --broker-list ip1:9092,ip1:9093,ip1:9094 --topic my-replicated-topic

2)从集群消费消息

./kafka-console-consumer.sh --bootstrap-server ip1:9092,ip1:9093,ip1:9094 --from-beginning --topic my-replicated-topic

3)指定消费组来消费消息

./kafka-console-consumer.sh --bootstrap-server ip1:9092,ip1:9093,ip1:9094 --from-beginning --consumer-property group.id=testGroup1 --topic my-replicated-topic

4)分区分消费组的集群消费中的细节

在这里插入图片描述

  • 一个partition只能被一个消费组中的一个消费者消费,目的是为了保证消费的顺序性,但是多个partition的多个消费者消费的总的顺序性是得不到保证的,那怎么做到消费的总顺序性呢?
  • partition的数量决定了消费组中消费者的数量,建议同一个消费组中消费者的数量不要超过partition的数量,否则多个消费者没有消息消费。
  • 如果消费者挂了,那么会触发rebalance机制,会让其他消费者来消费该分区。

文章来源:https://blog.csdn.net/qq_42236791/article/details/135270651
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。