Kafka 整体架构及概念_app

一、角色介绍：

Producer：消息生产者，负责发送消息到kafka集群 Broker : kafka服务端，负责接收客户端（producer/consumer ）请求、存储消息、协调消费组等。Broker没有主从之分，只不过通过Zookeeper选择出一个控制节点，控制节点负责管理整个集群中所有分区和副本的状态，例如分区故障后重新选择leader分区等。Consumer：负责从Broker上拉取消息进行业务处理。Zookeeper：负责协调整个Broker集群，负责选取控制Broker、存储集群元数据、topic路由信息存储等。Log：磁盘上的日志目录，消息就是存储在log中。Topic：kafka中的消息以topic为单位进行划分，生产者将消息发送到特定topic，而消费者负责订阅topic的消息并进行消费。Offset：offset是消息在分区中的唯一标识，kafka通过它来保证消息在分区的顺序性，不过offset并不跨越分区，也就是kafka保证的是分区有序性而不是主题有序性。Record：实际写入kafka中并可以被读取的消息记录。每个record包含了key、value和timestamp。Leader：每个分区多个副本的“主”leader，生产者发送数据的对象，以及消费者消费数据的对象都是leader。Follower：每个分区多副本中的“从”follower，实时从Leader中同步数据，保持和leader数据的同步。Leader发生故障时，某个follower会成为新的leader。Consumer Group：Consumer Group 为消费者组，一个消费者组可以包含一个或多个Consumer。使用多分区 + 多消费者方式可以极大提高数据下游的处理速度，同一消费者组中的消息不会重复消费，同样的，不同消费组中组中的消费者消费消息时互不影响。Kafka就是通过消费者组的方式来实现P2P模式和广播模式。Replication：副本，是kafka保证数据高可用的方式，kafka同一Partition的数据可以在多Broker上存在多个副本，通常只有主副本对外提供读写服务，当主副本所在broker崩溃或发生网络异常，kafka会在Controller的管理下重新选择新的Leader副本对外提供读写服务。Partition：一个topic可以分为多个分区，每个分区只属于单个主题。同一个主题下不同分区包含的消息是不同的，分区在存储层面可以看作一个可追加的日志（Log）文件，消息在被追加到分区日志文件的时候都会分配一个特定的偏移量（offset）。

producer(生产者)粗略

package com.bw.kfk;

import com.bw.util.KafkaUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

public class ProducerKfk {
    public static void main(String[] args) throws Exception {
        // TODO:  创建Flink流式处理环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        // TODO:  设置并行度为1
        environment.setParallelism(1);
        // TODO:  获取本地文件发送kfk
        DataStreamSource streamSource = environment.readTextFile( "src\\main\\resources\\data.txt" );
        streamSource.addSink( new FlinkKafkaProducer( "test",new SimpleStringSchema(),new KafkaUtils().producer() ) );

        streamSource .print(">>>");
        environment.execute();
    }
}

consumer(消费者)粗略

package com.bw.kfk;

import com.bw.bean.CountSum;
import com.bw.util.KafkaUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

public class ConsumerKfk {
    public static void main(String[] args) throws Exception {
        // TODO:  创建Flink流式处理环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        // TODO:  设置并行度为1
        environment.setParallelism(1);
        // TODO:  读取kfk数据
        DataStreamSource streamSource = environment.addSource(
                new FlinkKafkaConsumer( "test", new SimpleStringSchema(), new KafkaUtils().consummer() )
        );
        // TODO: 转换成java对象
        SingleOutputStreamOperator map = streamSource.map( new MapFunction() {
            @Override
            public CountSum map(String s) throws Exception {
                String[] data = s.split( "," );
                return new CountSum(
                        data[0],
                        data[1],
                        data[2]
                );
            }
        } );

        map.print(">>>");


        environment.execute();
    }
}

二、Kafka是如何做到高吞吐性能的

磁盘顺序读写：保证了消息的堆积顺序读写，零拷贝：避免CPU将数据从-块存储拷贝到另外-块存储的技术零拷贝：磁盘文件-->内核空间读取缓冲区-->网卡接口-->消费者进程分区分段 + 索引 kafka中的topic中的内容可以被分为多个partition存在，每个partition有分为多个段segment，所以每次 *** 作都是针对一小部分做 *** 作，很轻便，并且增加并行 *** 作的能力。批量压缩：多条消息一起压缩，降低宽带采用gzip的形式将消息进行压缩，多条消息一起压缩发送，降低宽带。同时对消息进行堆积，达到一定数量然后打个包进行发送。消费端批量拉取，进行解压。页面缓存技术

三、Kafka是如何实现消息有序的？有哪些分区算法？

Kafka是如何让实现消息有序的生产者：通过分区的leader副本负责数据以先进先出的顺序写入，来保证消息顺序行。消费者：同一分区内的消息只能被一个group里的消费者消费，保证分区内消费有序。Kafka每个partition中的消息写入时都是有序的，消费时，每个partition只能被每一个消费者组中的一个消费者消费，保证了消费时也是有序的。整个kafka不保证有序。如果为了保证kafka全局有序，那么设置一个生产者，一个分区，一个消费者。Kafka有哪些分区算法？轮询策略：将消息顺序分配到各个分区中，假设一个主题下有三个分区，第一条消息将被发送到分区0，第二条消息被发送到分区1，第三条消息被发送到分区2，以此类推，第四条消息被发送到分区0。轮询策略有非常优秀的负载均衡表现，它总是能保证消息最大限度地被平均分配到所有分区上，故默认情况下它是最合理的分区策略，也是我们最常用的分区策略之一。随机策略：将消息随机地放到任意一个分区上，本质上看随机策略也是力求将数据均匀地打散到各个分区，但从实际表现来看，它要逊色于轮询策略，所以如果追求数据的均匀分布，还是使用轮询策略比较好。事实上，随机策略是老版本生产者使用的分区策略，在新版本中已经改为轮询策略了按消息键保序策略：Kafka允许为每条消息创建消息建，称为key，每个key可以代表具体的业务含义，如业务ID，用户ID等。一旦消息被定义了key，那么你就可以保证同一个key的所有消息都进入到相同的分区里面，由于每个分区下的消息处理都是有顺序额，故这个策略被称为按消息键保序策略。

四、简单阐述kafka的ISR机制？如何保证每个应用程序都可以获取到kafka主题中的所有消息，而不是部分消息？

Kafka的ISR(In Sync Replicas)【ISR为同步副本】机制?

在分区中，所有副本统称为AR（Assigned Replicas）【AR为指定副本】，Leader维护了一个动态的ISR，ISR是指与leader副本保持同步状态的副本集合。当然leader副本本身也是集合中的一员。当ISR中的follower完成数据同步之后，leader就会给follower发送ack，如果其中一个follower长时间未向leader同步数据，该follower将会被踢出ISR集合，该时间阈值由replica.log.time.max.ms参数设定。当leader发生故障后，就会从ISR集合中重新选择出洗的leader。

五、kafka如何保证对应类型的消息被写到相同的分区？

通过消息建和分区器来实现，分区器为键生成一个offset，然后使用offset对主题分区进行取模，为消息选取分区，这样就可以保证包含同一个键的消息会被写到同一个分区上。

如果ProducerRecord没有指定分区，且消息的key不为空，则使用Hash算法（非加密型Hash函数，具备高运算性能及低碰撞率）来计算分区分配。如果ProducerRecord没有指定分区，且消息key也是空，则用轮询的方式选择一个分区。

如何保证每个程序都可以获取到Kafka主题中的所有消息，而不是部分消息？

1.在主题中创建消费者组，然后在往组中添加消费者来实现伸缩读取能力和处理能力，每个消费者组消费主题中的消息时，就互不干扰。

七、Kafka消费支持几种消费模式？如何保证数据的不重复和不丢失？Kafka是如何清理过期数据的？

Kafka支持几种消费者模式？ Kafka消费消息时支持三种模式： at most once 模式最多一次，可能会丢失但不会重复。at least once 模式至少一次，消息不会丢失但可能会重复。exactly once 模式精确传递一次。将offset作为唯一id与消息同时处理，并且保证处理的原子性。消息只会处理一次，不会丢失也不会重复。kafka默认模式是at least once，但这种模式可能会产生重复消息的问题，所以在业务逻辑必须做幂等设计。

如何保证数据的不重复和不丢失？ exactly once 模式精确传递一次。将offset作为唯一id与消息同时处理，并且保证处理的原子性。消息只会处理一次，不会丢失也不会重复。但这种方式很难做到。 kafka默认模式是at least once，但这种模式可能会产生重复消息的问题，所以在业务逻辑必须做幂等设计。使用 exactly once + 幂等 *** 作，可以保证数据不重复、不丢失。

Kafak是如何处理过期数据的？ Kafak将数据持久化到硬盘上，允许配置一定的策略对数据进行清洗，清理的策略有两个，删除和压缩。

数据清理的方式

1、删除： log.cleanup.policy=delete 启用删除策略直接删除,删除后的消息不可恢复。可以配置两个策略：#清理超过指定时间清理：log.retention.hours=16 #超过指定大小后，删除旧的消息：log.retention.bytes=1073741824 为了避免在删除时阻塞读 *** 作，采用了copy-on-write形式来实现，删除 *** 作进行时，读取 *** 作的copy on write ArrayList。

2、压缩：将数据压缩，只保留每个key最后一个版本的数据。首先在broker的配置中设置log.cleanup.enable=true启用cleaner，这个默认是关闭的。在topic的配置中设置log.cleanup.policy=compact 启用压缩策略。

八、====ACK机制Kafka的ack机制

文章目Kafka的ack机制录

前言

一、ACK的三个可选值

1、ACK = 1 （默认）

2、ACK = 0

3、ACK = -1

①、前言

Kafka的ACK机制，指的是producer（生产者）的消息发送确认机制，这直接影响到Kafka集群的吞吐量和消息可靠性。而吞吐量和可靠性就像硬币的两面，两者不可兼得，只能平衡。

Kafka的ACK机制是针对producer的。

保证消息的吞吐量就是producer只负责发数据，不需要得知消息是否被集群接收，这样就会有大量的数据发送到Kafka的集群中，保证了Kafka集群的吞吐量。