Kafka_02_Producer详解

Producer

Producer(生产者): 生产并发送消息到Broker(推送)

1. Producer是多线程安全的(建议通过池化以提高性能)
2. Producer实例后可发送多条消息(可对应多个ProducerRecord)

// 0.9之后的版本是基于Java实现(之前是Scala实现)

Producer客户端发送消息大致逻辑:

1. 配置Producer客户端参数并创建该Producer实例
2. 构建需发送的消息
3. 发送构建的消息
4. 关闭实例

构造Producer必填的3个参数:

参数	说明
bootstrap.servers	引导程序的服务地址 格式: 地址1:端口1,地址N:端口N (建议指定两个以上的Broker地址以保证稳定性, 且使用主机名形式)
key.serializer	发送时对Key调用的序列化器 Broker仅能接受字节数组形式的消息byte[]
value.serializer	发送时对Value调用的序列化器 Broker仅能接受字节数组形式的消息byte[]

// 序列化器必须以全限定名方式指定, Java的ProducerConfig类中包含所有的配置参数

ProducerRecord

ProducerRecord(构建消息): Producer每次发送的消息体

1. ProducerRecord由多个属性构成(Topic和消息是基础属性)
2. ProducerRecord有多个构造方法(指定属性的个数)
3. 可根据不同需求创建特定ProducerRecord

ProducerRecord定义:

public class ProducerRecord<K, 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;
    ......
}

Send&Close

Send(发送消息): Producer构建ProducerRecord之后发送给Broker

1. 发送模式: 发后既忘(fire-and-forget)、同步(sync)、异步(async)
2. 发送模式默认为异步(可通过获取返回值的方法以阻塞等待实现同步)
3. 返回值通常为发送消息的元数据(Topic、Partition、偏移量和时间戳等)

Send()方法的定义:

public Future<RecordMetadata> send(ProducerRecord<K, V> record);

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback);

1. 可通过Future的get()方法阻塞实现同步(返回RecordMetadata对象)
2. Send()方法需配合try/catch(发送成功或发生异常)
3. 发送导致的异常分为: 重试异常、不可重试异常

// 不可重试异常发生时会直接抛出并结束

常见的重试异常为：

可重试异常	说明
NewworkException	网络异常
LeaderNotAvailableException	副本的leader不可用 (可能正在选举leader)
UnknownTopicOrPartitionException	Topic或Partition异常
NotEnoughReplicasException	副本数量不足
NotCoordinatorException	协调器异常

Send()方法中的Callback定义:

public interface Callback {
    void onCompletion(RecordMetadata var1, Exception var2);
}

1. var1和var2参数互斥(两者必有个为null，后者代表异常)
2. 若两个消息对相同Partition发送消息, 则按发送顺序调用Callback

Close(结束发送):回收Producer实例

1. 发送结束后务必回收Producer实例(防止资源泄漏)
2. Close默认会阻塞等待之前所有的发送请求完成之后再回收
3. 可指定关闭的超时时间(超出该事件则强行回收, 不建议指定)

Close()方法的定义：

public void close();

public void close(long timeout, TimeUnit timeUnit);

实现原理

Producer的发送消息由两个线程完成:

1. 主线程: 构建并处理消息后发送至RecordAccumulator
2. Sender线程: 从RecordAccumulator获取消息, 并发送至Broker

如: Producer发送消息链路图

1. RecordAccumulator: 双端队列缓存待发送ProducerBatch以减少网络影响
2. ProducerBatch: 包含任意多个待发送的ProducerRecord(消息批次)
3. Request: Kafka支持的各种请求协议
4. InFlightRequests: 缓存已发送但未响应的Request

// Interceptor和Partitioner可选择性处理, 但必须经Serializer处理

Producer发送ProducerRecord的流程:

1. 主线程将ProducerRecord加工处理后发送至RecordAccumulator尾部
2. RecordAccumulator根据ProducerRecord分区选择对应的ProducerBatch
3. RecordAccumulator根据内存复用原则和ProducerBatch大小决定是否新建
4. Sender线程从RecordAccumulator头部获取ProducerBatch
5. 将<分区, <Deque<ProducerBatch>>形式变为<Node, List<ProducerBatch>>
6. 再根据各种协议请求转换为<Node, Request>形式
7. 发送前以Map<nodeId, Deque<Request>>缓存Request
8. 返回发送后的响应并清理InFlightRequests和RecordAccumulator

// 形式转换是为完成应用逻辑层到网络I/O层的转换

RecordAccumulator内存复用原则:

1. RecordAccumulator通过java.io.ByteBuffer和BufferPool实现内存复用
2. 若内存申请不超过指定大小, 则申请指定大小并放置于BufferPool
3. 若内存申请超过指定大小, 则申请该内存并再使用后直接释放

// BufferPool可避免频繁的申请和释放内存

InFlightRequest中包含leastLoadedNode

1. leastLoadedNode: 负载最小的Broker(未确认请求最少的)
2. leastLoadedNode常用于元数据请求和Consumer组播协议的交互
3. leastLoadedNode由Sender线程根据指定过期时间维护(主线程也可访问)

// 元数据: Broker、Topic、Partition、leader和follower副本所在的Broker等

如: Sender线程维护leatLoadedNode信息

1. Sender线程检查元数据是否过期(默认5m)
2. 超出则挑出leastLoadedNode, 向该Broker发送MetadataRequest请求
3. 获取结果后将其结果存入InFlightRequests中, 并更新元数据的过期时间

ProducerInterceptor

ProducerInterceptor(拦截器): 消息发送前/后的进行的操作

1. 不建议通过ProducerInterceptor修改topic、key和partition
2. 可指定多个ProducerInterceptor(拦截链按配置时顺序执行)
3. 可通过interceptor.classes参数指定Producer所使用的ProducerInterceptor

ProducerInterceptor定义:

public interface ProducerInterceptor<K, V> extends Configurable {
    // 发送前进行的操作
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);

    // 发送后被应答之后或失败进行的操作
    // 优先于Send()方法中定义的Callback前执行
    // 由于该方法运行于Producer的IO线程中, 应简洁
    public void onAcknowledgement(RecordMetadata metadata, Exception exception);

    // 关闭拦截器
    public void close();
}

// 抛出的任何异常都会被记录到日志中, 并不再向上抛

Serializer

Serializer(序列化器): 将特定数据转换成字节数组(byte[])

1. Broker仅能接受字节数组形式的数据(接收后会对其反序列化)
2. Producer使用的Serializer需和Consumer使用的反序列化器需对应
3. Producer指定Serializer时, 需通过全限定名方式指定(类的完整路径)

Serializer定义:

public interface Serializer<T> extends Closeable {
    // 配置序列化器
    // 常用于指定编码类型(默认UTF-8)
    void configure(Map<String, ?> configs, boolean isKey);

    // 执行序列化
    byte[] serialize(String topic, T data);

    // 关闭序列化器
    // 需保证幂等性
    void close();
}

// 不建议使用自定义Serializer或DeSerializer, 会增加耦合度

Partitioner

Partitioner(分区器): ProducerRecord分区的默认规则

1. ProducerRecord中指定partition字段, 则略过Partitioner
2. Partitioner的分区计算受Topic数量的影响(已分配的不受)
3. 可通过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(Map<String, ?> configs);
}

默认的Partitioner: org.apache.kafka.clients.producer.internals.DefaultPartitioner

1. 其close()方法默认为空
2. 消息为null时, 则以轮询的方式分配可用的分区号
3. 消息不为null时, 则进行Hash计算(MurmurHash2算法)

// 消息相同的情况下会写入相同的分区(存在消息互相覆盖的情况)

事务

事务(Transaction): Producer操作的最小原子单位(可跨Partition)

1. 开启事务时, 必须也需开启幂等性(enable.idempotence)
2. 开启事务时必须指定事务ID(若事务ID重复, 将结束被覆盖的事务并抛出异常)
3. 只能使事务处于以下两种状态(否则将抛出异常): COMMIT、ABORT
4. 事务开启后需关闭自动位移提交, 也不能位移消费

Producer中常用的事务方法:

// 初始化事务
void initTransactions();

// 开启事务
void beginTransaction();

// 事务内的位移提交
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId)

// 提交事务
void commitTransaction();

// 终止事务(回滚)
void abortTransaction();

事务协调器(TransactionCoordinator): 负责事务中的各类操作

1. 每个Producer都对应个事务协调器, 由其负责Producer中各类请求
2. 事务协调器会将事务的信息都存储至内部Toipc的__transaction_state

如: 事务的执行流程

1. 查找事务协调器: 找到事务协调器所在的Broker并建立连接(同时查找Partition)
2. 获取PID: 通过InitProducerIdRequest请求获取该事务ID
3. 执行事务: 通过各类请求处理Record并将数据存储至内部Topic
4. 结束事务: 发送各类请求结束事务, 同时将事务信息存储至内部Topic和日志文件

Consumer的事务受以下限制:

1. 采用日志压缩策略的Topic, 其Record可能被覆盖
2. Consumer在消费时可能没有分配到事务内的所有Partition
3. Record可能分布在Partition的多个LogSegment, 存在部分被清除的可能
4. Consumer可通过位移提交/位移消费访问Record, 可能导致遗漏事务中的Record