【Kafka消息队列】生产者发送消息流程

【Kafka消息队列】生产者发送消息流程

如何描述一条消息？

如何描述一条消息，就是在问这条消息的数据结构是什么？

public class ProducerRecord { private final String topic; //目标topic private final Integer partition; //目标partition private final Headers headers;//消息头信息 private final K key; //消息key private final V value; //消息体 private final Long timestamp; //消息时间戳 //构造方法，setter&&getter方法}

一条消息被抽象成ProducerRecord对象，其中topic表示主题，可以理解为一级目录。partition表示分区，属于主题下的二级目录。key表示路由的键，如果指定了key就相当于指定了分区。value表示消息体，最终的消息就是存在value中。

如何描述生产者？

Properties properties = new Properties(); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,“xxx:8080”); //broker的地址 //指定序列化器 properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer"); //指定过滤器 可配置多个用逗号隔开 properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"org.apache.kafka.clients.producer.SzzProducerInterceptorsTest"); //指定分区器（默认） properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"o.a.k.clients.producer.internals.DefaultPartitioner"); //构造 KafkaProducer KafkaProducer producer = new KafkaProducer(properties);

Properties 是一个map对象，只需要将配置放进propeties。KafkaProducer构造函数通过解析producer.propeties文件里面的属性来构造自己。例如 ：分区器、Key和Value序列化器、拦截器、RecordAccumulator消息累加器 、元信息更新器、启动发送请求的后台线程。

消息是怎么发送的？

拦截器

拦截器的执行在序列化和分区之前。

拦截器需要在Properties进行配置，配置后才能起到拦截作用。

序列化器

Producer需要将消息对象序列化成二进制数组，然后才能通过网络传输至Broker，消费者通过反序列化，将消息转换成消息对象。

Producer的序列化器和Consumer的反序列化器必须一致。否则解析会出问题。

分区器

分区器将消息进行路由，进而存储在相同topic下的不同分区，相同分区的消息是有序的，分区是实现负载均衡的一种策略，如下图所示。

分区策略

在消息对象数据结构中，有一个属性是partition，一旦指定了partition（也就是指定了分区），那么消息会不经过分区器直接将消息塞进指定的partition。

如果未指定partition，则按照key将消息路由至某个分区。至于是哪个分区，则要看使用的什么路由策略，常见的路由策略是轮询或者哈希。

如果key为null，则路由策略为轮询，将消息轮询至某个分区。如果key不为null，则路由策略为哈希，通过对key取哈希值，然后对partition的总数取模（该topic下的partition总数），取模后的值即是消息的要塞进的分区。

消息缓冲池

经过拦截器、分区器、序列化器以后的二进制消息并不是直接发送到broker，而是暂时放入消息缓冲池（recordAccumulator）中，

按照TopicPartition维度，创建队列（Deque），同理topic和Partition会被放在同一个队列里。

ProducerBatch 是队列中的元素结构，表示同一个批次发送的消息，大小为16k，能够存储一条或多条消息。

消息分流进缓冲池的流程

当一个消息进入缓冲池中，首先按照消息的TopicPartition找到指定的队列， 然后取队列中的最后一个元素（ProducerBatch），ProducerBatch的剩余内存能够塞下，则塞下这条消息，如果不能，则创建一个新的ProducerBatch。

Sender线程启动

sender是一个异步的IO线程，在获取Metadata时就已经唤醒。sender线程会循环轮询查看缓冲池中每个队列中ProducerBatch是否满足发送条件。sender将ProducerBatch发送后等待broker的response，然后调用回调函数。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。