分布式消息订阅-发布系统——Kafka

忘是亡心i 2022-11-26 11:00 268阅读 0赞

# 一、什么是Kafka #

![20200908092700115.jpg][]

Kafka是一个高吞吐量的分布式消息订阅-发布系统，其具备高性能、持久化、多副本备份、横向扩展能力。通过生产者向队列里写消息，消费者从队列里取消息进行业务逻辑，相对于传统的消息队列，实现业务逻辑的解耦，削峰和异步处理。

## 1、基本概念 ##

**主题（****Topic****）：**是特定类型的消息流。消息是字节的有效负载（Payload），话题是消息的分类名或种子（Feed）名。

**生产者（****Producer****）：**是能够发布消息到主题的任何对象。

**服务代理（****Broker****）：**已发布的消息保存在一组服务器中，它们被称为代理（Broker）或Kafka集群。

**消费者（****Consumer****）：**可以订阅一个或多个主题，并从Broker拉数据，从而消费这些已发布的消息。

**消费组（****Consumer Group****）：**多个消费者组成一个消费组，Kafka设计中同一个分区的消息只能被消费者组中的某一个消费者消费，同一个消费者组的消费者可以消费同一个Topic的不同分区消息。

**分区（****Partition****）：**Topic的分区，每个Topic可以有多个分区，用做负载均衡，提高Kafka的吞吐量。同一个Topic在不同的分区数据是不重复的，Partition表现形式就是一个一个文件夹。

**副本（****Replication****）：**每一个分区都有多个副本，作为数据备份。在Kafka中默认副本最大数量是10，且副本数不能大于Broker数。

**消息（****Message****）：**每一条发送的消息主体。

## 2、kafka的特性 ##

**高吞吐量、低延迟：**kafka每秒可以处理几十万条消息，延迟最低只有几毫秒，每个topic可以分为多个partition，consumer group对partition进行消费。

**可扩展性：**集群支持热扩展。

**持久性、可靠性：**消息被持久化到本地磁盘，并且支持数据备份防止数据丢失。

**容错性：**允许集群中节点失败（副本为n，允许失败n-1个节点）。

**高并发：**支持数千个客户端同时读写。

## 3、工作流程 ##

每个集群都由一个服务器作为leader，零或若干服务器作为follower，leader负责消息的读和写，follower负责数据备份，如果leader down了，follower中的一台通过zookeeper管理（注册一个临时节点）自动成为新leader。

![https://img-blog.csdnimg.cn/20200812085601290.png?x-oss-process=image/watermark,type\_ZmFuZ3poZW5naGVpdGk,shadow\_10,text\_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5,size\_16,color\_FFFFFF,t\_70][https_img-blog.csdnimg.cn_20200812085601290.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]

消息发送：

1.  producer先从集群获取分区的leader；
2.  producer将消息发送给leader；
3.  leader将消息写入本地文件；
4.  follower从leader pull消息（广播消息变化）；
5.  follower将消息写入本地后向leader发生ACK
6.  leader收到所有副本的Ack后向producer发生ACK。

![https://img-blog.csdnimg.cn/20200812085842959.png?x-oss-process=image/watermark,type\_ZmFuZ3poZW5naGVpdGk,shadow\_10,text\_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5,size\_16,color\_FFFFFF,t\_70][https_img-blog.csdnimg.cn_20200812085842959.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]

注：

1）follower主动去leader同步，每条数据追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的。

2） 一个topic通过多个partition，通过扩展机器轻松应对数据量增长，同时以partition为读写单位，可以多消费者同时消费，提升处理效率。

3）kafka将消息通过指定、hash值、轮询三种原则进行负载均衡。

4）利用ACK应答机制实现消息发生可靠性：“0”不需要应答、“1”leader应答、“all”所有应答。

![https://img-blog.csdnimg.cn/20200812085913738.png?x-oss-process=image/watermark,type\_ZmFuZ3poZW5naGVpdGk,shadow\_10,text\_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5,size\_16,color\_FFFFFF,t\_70][https_img-blog.csdnimg.cn_20200812085913738.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]

消息保存：

![https://img-blog.csdnimg.cn/20200812085859541.png?x-oss-process=image/watermark,type\_ZmFuZ3poZW5naGVpdGk,shadow\_10,text\_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5,size\_16,color\_FFFFFF,t\_70][https_img-blog.csdnimg.cn_20200812085859541.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]

一个Topic包含parition（服务器上表现为文件夹），一个partition包含多组segment，每个segment包含.index、.log、.timeindex三个文件。其中，log存储实际message，index和timeindex为索引文件，用于检索消息。无论消息是否被消费，kafka都会基于时间（7天）/大小（1GB）保存所有消息。

注：kafka读取特定消息的时间复杂度为O(1)，删除过期文件不会提高性能。

消息消费：

![https://img-blog.csdnimg.cn/20200812085824873.png?x-oss-process=image/watermark,type\_ZmFuZ3poZW5naGVpdGk,shadow\_10,text\_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5,size\_16,color\_FFFFFF,t\_70][https_img-blog.csdnimg.cn_20200812085824873.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]

1.  先找到offset为368801的message所在的segment（二分查找）；
2.  打开对应segment的.index文件，因为利用稀疏索引存储，所有同样利用二分查找到小于或等于offset的指定索引条目；
3.  根据相对index索引条目去确定存储的物理偏移位置，顺序查找指定消息。

注：利用segment+有序offset+稀疏索引\+二分查找\+顺序查找实现高效查找。

消息存储在log文件后，消费者就可以进行消费，多个消费者可以组成一个消费组，每个消费者组都有一个id，同一个消费组的消费者可以消费同一个topic下的不同分区，但一个分区只能一个消费者消费。当消费者<分区数，部分消费者消费多个分区；当消费者>分区数。部分消费者不消费。因此，建议消费者组的消费者数量=分区数量。

# 二、简单操作 #

根据使用场景的不同，Kafka提供了两种方式操作集群：（1）在Linux Shell环境下利用命令行对Kafka进行操作；（2）Kafka对多种编程语言提供的API调用，演示以Java API为例。

## **1、Kafka****的****shell****脚本操作：** ##

1、Kafka启动与关闭

启动：

bin/kafka-server-start.sh  ../config/server.properties &

关闭：

bin/kafka-server-stop.sh stop

注：如果在kafka的bin目录下执行，则命令可以以./相应命令，如：

./kafka-server-stop.sh stop

2、查看Kafka的topics

./kafka-topics.sh  --list --bootstrap-server service-kafka:9092

注：--bootstrap-server指目标集群服务器地址，在kafka0.8版本之前为--zookeeper

3、创建Kafka的topics

./kafka-topics.sh  --create --bootstrap-server service-kafka:9092  --topic testcyc  --partitions 3 --replication-factor 1

参数说明：

\--partitions 创建topic的分区数

\--replication-factor 创建topic的副本数

4、删除Kafka的topics

./kafka-topics.sh --delete --bootstrap-server service-kafka:9092 --topic testcyc

5、创建生产者

./kafka-console-producer.sh  --topic testcyc --broker-list service-kafka:9092

参数说明：

\--broker-liat 同理bootstrap-server指目标集群服务器地址

6、创建消费者

./kafka-console-consumer.sh  --bootstrap-server service-kafka:9092 --topic testcyc

## **2、Kafka的****Java API****操作：** ##

1、创建生产者

Properties properties = new Properties();
    //连接kafka集群
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.3.69.118:9092");
    //Key序列化
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
    //Value序列化
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    //创建生产者
    KafkaProducer<Integer,String> producer = new KafkaProducer<Integer, String>(properties);
    
    int num=0;
    while(num<10){
        String msg = "Kafka test"+num;
        try {
            producer.send(new ProducerRecord<Integer, String>("testkafka",msg)).get();
            TimeUnit.SECONDS.sleep(2);
            num++;
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

2、创建消费者

Properties properties = new Properties();
    //连接kafka集群
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.3.69.118:9092");
    //指定消费组
    properties.put(ConsumerConfig.GROUP_ID_CONFIG,"testkafka");
    //设置offfset自动提交
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");
    //自动提交间隔时间
    properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
    //Key反序列化
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
    //Value反序列化
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    
    //创建消费者
    KafkaConsumer<Integer,String> consumer = new KafkaConsumer<Integer, String>(properties);
    consumer.subscribe(Collections.singleton("testcyc"));
    while(true){
        ConsumerRecords<Integer,String> records = consumer.poll(Duration.ofSeconds(100));
        for(ConsumerRecord<Integer,String> record :records){
            System.out.println(record.key()+" "+record.value()+"->offset:"+record.offset());
        }
    }

[20200908092700115.jpg]: /images/20221124/072f15245da04dc3b6298c4bc42a186a.png
[https_img-blog.csdnimg.cn_20200812085601290.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]: /images/20221124/c04546d446ae44dca5741e845e0911fd.png
[https_img-blog.csdnimg.cn_20200812085842959.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]: /images/20221124/07739bbcd54c4537b70c7a0477f6f460.png
[https_img-blog.csdnimg.cn_20200812085913738.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]: /images/20221124/a11e5a24a20b4e67b27877d26015b9f4.png
[https_img-blog.csdnimg.cn_20200812085859541.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]: /images/20221124/d854b2991e3549d3b0ba40c2fb5e8d05.png
[https_img-blog.csdnimg.cn_20200812085824873.png_x-oss-process_image_watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI1MDYyMjk5_size_16_color_FFFFFF_t_70]: /images/20221124/d91b69f7d9cb45d8a3ed8da7e1b91496.png