分布式消息队列kafka【二】—— 基础概念介绍和快速入门

分布式消息队列kafka【二】—— 基础概念介绍和快速入门

Kafka介绍与高性能原因分析

Kafka介绍

• Kafka是LinkedIn开源的分布式消息系统，目前归属于Apache顶级项目
• Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输
• 0.8版本开始支持复制，不支持事务，目的追求高吞吐量。对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务
• Kafka可以做到消息的可靠性，比如一条不丢失，但这样对Kafka的性能有一定的降低，正常情况即然选择了Kafka就会容忍它极少的数据丢失，比如十几亿的日志收集丢了两三条，问题不大

Kafka有哪些特点

• 具有分布式特性，支持消息分区概念，核心概念partition(分区)，一个topic(主题)下可以有多个partition，partition和consumer(消费者)是一一对应的，每个topic的某一个partition只能被同一个消费组下的其中一个consumer消费，同组的consumer则起到均衡效果。因此我们可以说分区是消费并行度的基本单位。从consumer的角度讲，我们订阅消费了一个topic，也就订阅了该topic的所有partition。当消费者数量多于partition的数量时，多余的消费者空闲。多个消费者组，就是浪费的，无意义的。组与组之间的消息是否被消费是相互隔离互不影响的
• 跨平台，支持不同语言的客户端，比如java、php、python等，对于异构的系统使用友好
• 实时性，数据支持实时处理和一键处理，即使Kafka数据堆积上亿，只要存储OK，不会影响Kafka性能，不会影响Kafka消息的接收和发送性能
• 伸缩性，支持水平扩展

Kafka高性能的原因是什么？

• 顺序写，PageCache（空中接力，高效读写）
  • 顺序写就是将消息不断追加写入本地磁盘，随机写，是指可以在任何时候将存取文件的指针指向文件内容的任何位置。顺序写的性能是随机写的万倍。
  • PageCache详解见下
• 高性能、高吞吐，初衷，日志收集。
• 后台异步，主动Flush，好多异步级别的scheduler，将连续的小块组成一个大块的物理文件。文件按顺序排好，减少磁盘移动时间，充分利用空闲的内存。
• 预读策略，IO调度，zeroCopy

Kafka高性能核心pageCache与zeroCopy原理解析

PageCache

PageCache是系统级别的缓存，它把尽可能多的空闲内存当作磁盘缓存使用来进一步提高IO效率，同时当其他进程申请内存，回收PageCache的代价也很小。当上层有写操作时，操作系统只是将数据写入PageCache，同时标记Page属性为Dirty。当读操作发生时，先从PageCache中查找，如果发生缺页才进行磁盘调度，最终返回需要的数据。PageCache同时可以避免在JVM内部缓存数据，避免不必要的GC、以及内存空间占用。对于In-Process Cache，如果Kafka重启，它会失效，而操作系统管理的PageCache依然可以继续使用。

对应到Kafka生产和消费消息中

producer把消息发到broker后，数据并不是直接落入磁盘的，而是先进入PageCache。PageCache中的数据会被内核中的处理线程采用同步或异步的方式写回到磁盘。

Consumer消费消息时，会先从PageCache获取消息，获取不到才回去磁盘读取，并且会预读出一些相邻的块放入PageCache，以方便下一次读取。

如果Kafka producer的生产速率与consumer的消费速率相差不大，那么几乎只靠对broker PageCache的读写就能完成整个生产和消费过程，磁盘访问非常少。

传统应用程序读取磁盘文件和返回给消费者的过程

zeroCopy（零拷贝）

磁盘文件只在内核空间上下文copy一次到内核读取缓冲区，然后直接写到网卡给消费者。不同消费者可以使用同一个内核读取缓冲区的磁盘文件数据，避免重复复制。

Kafka(MQ)实战应用场景剖析

Kafka(MQ)的应用场景

• Kafka之异步化、服务解耦、削峰填谷
• Kafka海量日志收集
• Kafka之数据同步应用
• Kafka之实时计算分析

Kafka(MQ)之异步化实战

如图，假设线上服务出现BUG，有一个error日志告警，此时需要把这个error日志以短信、邮件、第三方推送等方式通知用户，如果都集中到单台服务器，压力过大。我们可以把这些推送封装为一个个请求，推送代理服务是一个生产者的多集群形式，需要持久化的请求持久化存储，将请求按照类型划分发送到具体处理的某个集群，比如是邮件consumer服务集群，推送请求经过某个推送代理服务集群，生产一条消息发送到邮件的topic中，邮件consumer服务集群的接收到topic中的消息处理请求并发送邮件，比如某段时间同时生产30个请求，邮件consumer服务集群有三个节点，可以做负载均衡，每个节点处理10个请求。从而实现了异步化的处理，还启到了服务解耦的作用。

Kafka(MQ)之服务解耦、削峰填谷

如图，实现了订单系统和物流系统的服务解耦，而且比如双十一订单量极具增加的情况下，用户更新订单状态后发送到MQ，可以通过延迟消费将处理不过来的消息延迟到并发量小的时间段消费，从而实现削峰填谷的作用。

Kafka海量日志收集

如图，图中左边的Log4j2 Appender是应用服务集群，会产生一个全量日志app.log和错误日志error.log。然后通过filebeat日志抓取插件，抓取这两个日志文件推送到kafka中。如果只推送全量日志，这就无法保证error日志的即时处理，同时kafka的topic也要分开。kafka的数据再流入下游logstash中，对日志进行解析操作，解析完的json数据再存储到es中。这就是一套比较完整的kafka海量日志收集系统。

Kafka(MQ)之数据同步实战

如图，比如订单的创建，订单的流量非常大时，我们持久化到数据库层，肯定需要一些分库分表的策略。怎么去做整体维度的统计分析和查询？连接这些数据库并行的查询，然后做结果的汇总吗，肯定不是这么去做的！我们需要把这些数据库的所有表的数据统一到一个地方进行查询。目前市面上比较主流的是同步到es、HBase、redis等。图中cannal是解析Mysql Binlog实时同步的中间件。Mysql Binlog怎么解析？需要开启Mysql Binlog，指定按照一行行的读取，读取之后发送到MQ。然后由consumer消费一条条的Binlog数据，解析Binlog数据同步到es上。这就是一套比较完整的kafka数据同步系统。

另外，还有的公司采用这种方式实现数据同步：有些数据同步是直接在订单创建的时候，先入库后再发到MQ，MQ再将数据同步到es中。这种方式其实是不可靠的，无法保证入库后再发到MQ的原子性，这种就需要实现像RabbitMQ的可靠性投递。最可靠的是已经落库的数据，从这个源头去同步更为简单。比如图中这种方式。

Kafka之实时计算分析

如图，数据采集一些application应用，发送到kafka，kafka直接对接Flink实时计算平台，Flink对数据进行一些统计分析、聚合，然后再把数据输出到Destination。

kafka基础概念

kafka集群架构

中间Kafka broker代表三个Kafka节点，奇数个稳定。Producer会将消息发送到Kafka集群，Consumer会拉取Kafka集群消息到本地处理。Kafka集群的维护通过元数据管理中心Zookeeper实现。

Kafka集群大部分都是内存级别的存储，如果Producer的生产速率与Consumer的消费速率相差不大，甚至完全不需要磁盘，此时磁盘只做备份的作用，replicate表示数据同步，是基于内存级别的，假设此时有三个节点，相同的一份消息在内存里存了三份，当某一两个节点宕机，不会有数据丢失，但是三个节点都宕机，会有部分数据丢失，当然，kafka也有相关的配置保证数据不丢失，但是会影响性能。

topic与partition关系

Topic：主题（逻辑概念），Partition：分区（可以简单的理解为物理概念，在存储层面上可以理解为一个个可以追加的日志文件）。如图，TopicA由1、2、3、4四个分区，注意：一个分区只能属于单个主题，同一个主题下可以由多个分区。消息被追加到分区里，都会临时分配偏移量（offset）。偏移量（offset）在分区中是一个唯一的标识。对于这个偏移量（offset）的追加方式是顺序写的过程。Kafka可以保证分区是有序的，不是主题是有序的。

怎么保证消息的有序？只需要保证一个消费者/一个线程去连接一个分区即可。

每一条消息在发送到kafka broker之前，会根据一个分区的规则（分区器），通过这个规则（可能是路由规则、hash规则等）把消息变成0、1、2、3这样的分区序号，然后把消息打到指定的分区上。

如果某个主题上只有一个分区，即一个磁盘文件，那所有操作的IO都会产生在这一个磁盘文件上，会产生资源的瓶颈。分区的目的就是为了分散磁盘的IO。

创建主题时，可以指定分区数，当然后续也可以修改某个主题的分区数。通过增加分区可以更好的实现水平扩展性。

副本概念（replica）

Broker代表实际的物理机器节点，绿色的P代表leader副本，紫色的P代表follower副本。可以通过增加follower副本的数量提升集群的容灾能力，比如这里Broker3宕机，还可以从Borker1和2拼接1、2、3、4四个副本，保证数据不丢失。

每个partition可以有多个副本。

• leader副本：每个partition只能有一个leader副本，所有的producer的请求都会发送到这个leader 副本，当然consumer可以从leader上consume数据也可以从follower replica中consume数据。
• follower副本：partition中除了leader副本之外的副本就称之为follower副本，follower副本的数目可以自由配置，不像leader只能有一个。它的主要任务就是保持和leader同步，当旧的leader副本出现问题的时候它能够快速被晋级为leader副本，从而保证高可用性。我们可以从follower副本中读，也可以只从leader读，这个是可以配置的。

需要注意的是：同一时刻，leader副本和follower副本的数据并非完全一致，之间存在主从概念，会有数据同步的过程。绿色的leader副本可以处理读写请求，紫色的follower副本只负责去实时拉取绿色的leader副本的数据，进行数据同步。

In Sync Replicas

AR：Assigned Replicas，分区中的所有副本统称为AR

ISR：In Sync Replicas，所有与leader副本保持一定程度同步的副本（包括leader副本在内）组成ISR

OSR：Out Sync Relipcas，与leader副本同步滞后过多的副本（不包括leader副本），组成OSR

由此可见：AR=ISR+OSR。在正常情况下，所有的follower副本都应该与leader副本保持一定程度的同步，即AR=ISR,OSR集合为空。

图中绿色的P1代表leader副本，紫色的P1 S1和P1 S2代表follower副本，当外部请求写入消息到P1 leader副本后，follower副本S1和S2还需要拉取P1数据并写入。

leader副本主要维护和跟踪ISR集合里所有follower副本的滞后状态。当OSR集合中的follower副本同步的进度跟上了，会把OSR集合中的follower副本写回ISR集合。

当leader副本所在的节点宕机，只会从ISR集合中选取follower副本成为leader副本。

高水位线

• HW：High Watermark，高水位线，消费者只能最多拉取到高水位线的消息
• LEO：Log End Offset，日志文件的最后一条记录的offset（偏移量）
• ISR集合与HW和LEO存在着密不可分的关系

如图，HW在6的位置，故消费者最多可消费的消息区间是在0-5之间，不能消费6-8的数据，LEO代表最新的日志文件写入在9的位置。

HW，高水位线，代表消费者只能最多拉取到高水位线的消息。leader副本在写入消息后，follow副本还没有成功拉取消息，所以此时消息还无法消费。

follow1同步数据很快，follow2同步3消息很快，但是同步4消息很慢，此时HW会一直在3这个位置上。

当然这个可以通过选择策略改变，可以选择一条消息发送出去后，所有副本必须都同步成功才能返回结果，才能拉取这个offset消息消费。也可以选择半数以上同步成功即可拉取这个消息消费。

kafka快速入门

生产者编码

操作步骤

• 配置生产者参数属性和创建生产者对象
• 构建消息：ProducerRecord
• 发送消息：send
• 关闭生产者

消息实体（后续复用）

packagecom.bfxy.kafka.api.model;importlombok.AllArgsConstructor;importlombok.Data;importlombok.NoArgsConstructor;importjava.io.Serializable;@Data@AllArgsConstructor@NoArgsConstructorpublicclassUser/*implements Serializable*/{// 不需要实现Serializable接口，因为kafka对VALUE已经做了序列化privateStringid;privateStringname;}

topic常量

packagecom.bfxy.kafka.api.constant;publicinterfaceConst{StringTOPIC_QUICKSTART="topic_quickstart";}

生产者

packagecom.bfxy.kafka.api.quickstart;importcom.alibaba.fastjson.JSON;importcom.bfxy.kafka.api.constant.Const;importcom.bfxy.kafka.api.model.User;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.clients.producer.ProducerRecord;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;publicclassQuickStartProducer{publicstaticvoidmain(String[]args){// 1.配置生产者启动的关键属性参数Propertiesproperties=newProperties();// 1.1.连接kafka集群的服务列表，如果有多个，使用”逗号“隔开properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.218.21:9092");// 1.2.CLIENT_ID_CONFIG：这个属性的目的是标记kafkaClient的IDproperties.put(ProducerConfig.CLIENT_ID_CONFIG,"quickstart-producer");// 1.3.对kafka的key以及value做序列化，KEY_SERIALIZER_CLASS_CONFIG和VALUE_SERIALIZER_CLASS_CONFIG// Q：为什么需要对kafka的key以及value做序列化？// A：因为Kafka Broker在接收消息的时候，必须要以二进制的方式接收，所以必须要对key以及value做序列化// 字符串序列化类（可以直接用后面这串字符串）：org.apache.kafka.common.serialization.StringSerializer// KEY：是kafka用于做消息投递计算具体投递到对应主题的哪一个分区而需要的properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// VALUE：实际发送消息的内容properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// try的括号种进行创建资源连接的写法的作用：在try语句结束后自动释放，前提是这些可关闭的资源必须实现 java.lang.AutoCloseable 接口。此时，就不用再finally中进行资源的释放了。// 2.创建kafka生产者对象，传递properties属性参数集合try(KafkaProducer<String,String>producer=newKafkaProducer<>(properties)){for(inti=0;i<10;i++){// 3.构造消息内容Useruser=newUser("00"+i,"张三");// arg1：topic，arg2：实际的消息体内容ProducerRecord<String,String>record=newProducerRecord<>(Const.TOPIC_QUICKSTART,JSON.toJSONString(user));// 4.发送消息producer.send(record);}}catch(Exceptione){e.printStackTrace();}}}

消费者编码

操作步骤

• 配置消费者参数属性和构造消费者对象
• 订阅主题
• 拉取消息并进行消费处理
• 提交消费偏移量，关闭消费者

消费者

packagecom.bfxy.kafka.api.quickstart;importcom.bfxy.kafka.api.constant.Const;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importorg.apache.kafka.common.TopicPartition;importorg.apache.kafka.common.serialization.StringDeserializer;importjava.time.Duration;importjava.util.Collections;importjava.util.List;importjava.util.Properties;publicclassQuickStartConsumer{publicstaticvoidmain(String[]args){// 1.配置属性参数Propertiesproperties=newProperties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.218.21:9092");// 字符串反序列化类（可以直接用后面这串字符串）：org.apache.kafka.common.serialization.StringDeserializerproperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());// 非常重要的属性配置：与消费者订阅组有关系properties.put(ConsumerConfig.GROUP_ID_CONFIG,"quickstart-group");// 常规属性：会话连接超时时间，默认是10000properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,10000);// 消费者提交offset：自动提交&手动提交，默认是自动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);// 自动提交的提交周期，默认5000properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,5000);// 2.创建消费者对象try(KafkaConsumer<String,String>consumer=newKafkaConsumer<>(properties)){// 3.订阅感兴趣的主题consumer.subscribe(Collections.singletonList(Const.TOPIC_QUICKSTART));System.out.println("quickstart consumer started...");// 4.采用拉取消息的方式消费数据while(true){// 设置等待多久拉取一次消息ConsumerRecords<String,String>records=consumer.poll(Duration.ofMillis(1000));// 拉取Const.TOPIC_QUICKSTART主题里面所有的消息// topic和partition是一对多的关系，一个topic可以有多个partition// 因为消息是在partition里存储的，所以需要遍历partition集合for(TopicPartitionpartition:records.partitions()){// 通过TopicPartition获取指定的消息集合，获取到的就是当前TopicPartition里面所有的消息List<ConsumerRecord<String,String>>partitionRecordList=records.records(partition);// 获取当前TopicPartition对应的主题名称Stringtopic=partition.topic();// 获取当前TopicPartition下的消息条数intsize=partitionRecordList.size();System.out.println(String.format("--- 获取topic：%s，分区位置：%s，消息总数：%s ---",topic,partition.partition(),size));for(ConsumerRecord<String,String>consumerRecord:partitionRecordList){// 实际的消息内容Stringvalue=consumerRecord.value();// 当前获取的消息偏移量longoffset=consumerRecord.offset();// 提交的消息偏移量：如果要提交的话，必须提交当前消息的offset+1，表示下一次从什么位置（offset）拉取消息longcommitOffset=offset+1;System.out.println(String.format("-- 获取实际消息value：%s，消息offset：%s，提交offset：%s ---",value,offset,commitOffset));}}}}catch(Exceptione){e.printStackTrace();}}}