烟台建设科技网站,超市网站源码,什么专业会做网站,凡客登录入口Kafka_02_Producer详解 ProducerProducerRecordSendClose实现原理ProducerInterceptorSerializerPartitioner 事务 Producer
Producer(生产者): 生产并发送消息到Broker(推送)
Producer是多线程安全的(建议通过池化以提高性能)Producer实例后可发送多条消息(可对应多个P… Kafka_02_Producer详解 ProducerProducerRecordSendClose实现原理ProducerInterceptorSerializerPartitioner 事务 Producer
Producer(生产者): 生产并发送消息到Broker(推送)
Producer是多线程安全的(建议通过池化以提高性能)Producer实例后可发送多条消息(可对应多个ProducerRecord)
// 0.9之后的版本是基于Java实现(之前是Scala实现) Producer客户端发送消息大致逻辑:
配置Producer客户端参数并创建该Producer实例构建需发送的消息发送构建的消息关闭实例 构造Producer必填的3个参数:
参数说明bootstrap.servers引导程序的服务地址格式: 地址1:端口1,地址N:端口N(建议指定两个以上的Broker地址以保证稳定性, 且使用主机名形式)key.serializer发送时对Key调用的序列化器Broker仅能接受字节数组形式的消息byte[]value.serializer发送时对Value调用的序列化器Broker仅能接受字节数组形式的消息byte[]
// 序列化器必须以全限定名方式指定, Java的ProducerConfig类中包含所有的配置参数 ProducerRecord
ProducerRecord(构建消息): Producer每次发送的消息体
ProducerRecord由多个属性构成(Topic和消息是基础属性)ProducerRecord有多个构造方法(指定属性的个数)可根据不同需求创建特定ProducerRecord ProducerRecord定义:
public class ProducerRecordK, V {private final String topic; // Topic(必填)private final Integer partition; // Partition// 消息头部(0.11版本引入)// 指定与应用相关信息(可忽略)private final Headers headers;// 键(附加信息)// 其会用于计算Partition(二次归类)private final K key;// 值(消息体, 必填)// 为空则代表: 墓碑消息private final V value;// 消息时间戳// 细分为CreateTime(消息创建时间)和LogAppendTime(追加日志时间)private final Long timestamp;......
}SendClose
Send(发送消息): Producer构建ProducerRecord之后发送给Broker
发送模式: 发后既忘(fire-and-forget)、同步(sync)、异步(async)发送模式默认为异步(可通过获取返回值的方法以阻塞等待实现同步)返回值通常为发送消息的元数据(Topic、Partition、偏移量和时间戳等) Send()方法的定义:
public FutureRecordMetadata send(ProducerRecordK, V record);public FutureRecordMetadata send(ProducerRecordK, V record, Callback callback);可通过Future的get()方法阻塞实现同步(返回RecordMetadata对象)Send()方法需配合try/catch(发送成功或发生异常)发送导致的异常分为: 重试异常、不可重试异常
// 不可重试异常发生时会直接抛出并结束 常见的重试异常为
可重试异常说明NewworkException网络异常LeaderNotAvailableException副本的leader不可用(可能正在选举leader)UnknownTopicOrPartitionExceptionTopic或Partition异常NotEnoughReplicasException副本数量不足NotCoordinatorException协调器异常 Send()方法中的Callback定义:
public interface Callback {void onCompletion(RecordMetadata var1, Exception var2);
}var1和var2参数互斥(两者必有个为null后者代表异常)若两个消息对相同Partition发送消息, 则按发送顺序调用Callback Close(结束发送):回收Producer实例
发送结束后务必回收Producer实例(防止资源泄漏)Close默认会阻塞等待之前所有的发送请求完成之后再回收可指定关闭的超时时间(超出该事件则强行回收, 不建议指定) Close()方法的定义
public void close();public void close(long timeout, TimeUnit timeUnit);实现原理
Producer的发送消息由两个线程完成:
主线程: 构建并处理消息后发送至RecordAccumulatorSender线程: 从RecordAccumulator获取消息, 并发送至Broker 如: Producer发送消息链路图 RecordAccumulator: 双端队列缓存待发送ProducerBatch以减少网络影响ProducerBatch: 包含任意多个待发送的ProducerRecord(消息批次)Request: Kafka支持的各种请求协议InFlightRequests: 缓存已发送但未响应的Request
// Interceptor和Partitioner可选择性处理, 但必须经Serializer处理 Producer发送ProducerRecord的流程:
主线程将ProducerRecord加工处理后发送至RecordAccumulator尾部RecordAccumulator根据ProducerRecord分区选择对应的ProducerBatchRecordAccumulator根据内存复用原则和ProducerBatch大小决定是否新建Sender线程从RecordAccumulator头部获取ProducerBatch将分区, DequeProducerBatch形式变为Node, ListProducerBatch再根据各种协议请求转换为Node, Request形式发送前以MapnodeId, DequeRequest缓存Request返回发送后的响应并清理InFlightRequests和RecordAccumulator
// 形式转换是为完成应用逻辑层到网络I/O层的转换 RecordAccumulator内存复用原则:
RecordAccumulator通过java.io.ByteBuffer和BufferPool实现内存复用若内存申请不超过指定大小, 则申请指定大小并放置于BufferPool若内存申请超过指定大小, 则申请该内存并再使用后直接释放
// BufferPool可避免频繁的申请和释放内存 InFlightRequest中包含leastLoadedNode
leastLoadedNode: 负载最小的Broker(未确认请求最少的)leastLoadedNode常用于元数据请求和Consumer组播协议的交互leastLoadedNode由Sender线程根据指定过期时间维护(主线程也可访问)
// 元数据: Broker、Topic、Partition、leader和follower副本所在的Broker等 如: Sender线程维护leatLoadedNode信息
Sender线程检查元数据是否过期(默认5m)超出则挑出leastLoadedNode, 向该Broker发送MetadataRequest请求获取结果后将其结果存入InFlightRequests中, 并更新元数据的过期时间 ProducerInterceptor
ProducerInterceptor(拦截器): 消息发送前/后的进行的操作
不建议通过ProducerInterceptor修改topic、key和partition可指定多个ProducerInterceptor(拦截链按配置时顺序执行)可通过interceptor.classes参数指定Producer所使用的ProducerInterceptor ProducerInterceptor定义:
public interface ProducerInterceptorK, V extends Configurable {// 发送前进行的操作public ProducerRecordK, V onSend(ProducerRecordK, V record);// 发送后被应答之后或失败进行的操作// 优先于Send()方法中定义的Callback前执行// 由于该方法运行于Producer的IO线程中, 应简洁public void onAcknowledgement(RecordMetadata metadata, Exception exception);// 关闭拦截器public void close();
}// 抛出的任何异常都会被记录到日志中, 并不再向上抛 Serializer
Serializer(序列化器): 将特定数据转换成字节数组(byte[])
Broker仅能接受字节数组形式的数据(接收后会对其反序列化)Producer使用的Serializer需和Consumer使用的反序列化器需对应Producer指定Serializer时, 需通过全限定名方式指定(类的完整路径) Serializer定义:
public interface SerializerT extends Closeable {// 配置序列化器// 常用于指定编码类型(默认UTF-8)void configure(MapString, ? configs, boolean isKey);// 执行序列化byte[] serialize(String topic, T data);// 关闭序列化器// 需保证幂等性void close();
}// 不建议使用自定义Serializer或DeSerializer, 会增加耦合度 Partitioner
Partitioner(分区器): ProducerRecord分区的默认规则
ProducerRecord中指定partition字段, 则略过PartitionerPartitioner的分区计算受Topic数量的影响(已分配的不受)可通过partitioner.class参数指定Producer所使用的Partitioner Partitioner定义:
public interface Partitioner extends Configurable, Closeable {// 计算并返回分区号public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);// 关闭分区器public void close();
}public interface Configurable {// 获取配置信息并初始化数据void configure(MapString, ? configs);
}默认的Partitioner: org.apache.kafka.clients.producer.internals.DefaultPartitioner
其close()方法默认为空消息为null时, 则以轮询的方式分配可用的分区号消息不为null时, 则进行Hash计算(MurmurHash2算法)
// 消息相同的情况下会写入相同的分区(存在消息互相覆盖的情况) 事务
事务(Transaction): Producer操作的最小原子单位(可跨Partition)
开启事务时, 必须也需开启幂等性(enable.idempotence)开启事务时必须指定事务ID(若事务ID重复, 将结束被覆盖的事务并抛出异常)只能使事务处于以下两种状态(否则将抛出异常): COMMIT、ABORT事务开启后需关闭自动位移提交, 也不能位移消费 Producer中常用的事务方法:
// 初始化事务
void initTransactions();// 开启事务
void beginTransaction();// 事务内的位移提交
void sendOffsetsToTransaction(MapTopicPartition, OffsetAndMetadata offsets, String consumerGroupId)// 提交事务
void commitTransaction();// 终止事务(回滚)
void abortTransaction();事务协调器(TransactionCoordinator): 负责事务中的各类操作
每个Producer都对应个事务协调器, 由其负责Producer中各类请求事务协调器会将事务的信息都存储至内部Toipc的__transaction_state 如: 事务的执行流程 查找事务协调器: 找到事务协调器所在的Broker并建立连接(同时查找Partition)获取PID: 通过InitProducerIdRequest请求获取该事务ID执行事务: 通过各类请求处理Record并将数据存储至内部Topic结束事务: 发送各类请求结束事务, 同时将事务信息存储至内部Topic和日志文件 Consumer的事务受以下限制:
采用日志压缩策略的Topic, 其Record可能被覆盖Consumer在消费时可能没有分配到事务内的所有PartitionRecord可能分布在Partition的多个LogSegment, 存在部分被清除的可能Consumer可通过位移提交/位移消费访问Record, 可能导致遗漏事务中的Record
