Kafka_02_Producer详解-Toy模板网

这篇具有很好参考价值的文章主要介绍了Kafka_02_Producer详解。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

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 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

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

Send()方法的定义:

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

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

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

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

常见的重试异常为：

可重试异常	说明
NewworkException	网络异常
LeaderNotAvailableException	副本的leader不可用 (可能正在选举leader)
UnknownTopicOrPartitionException	Topic或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的发送消息由两个线程完成:

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

如: Producer发送消息链路图

Kafka_02_Producer详解,Kafka,互联网精神,kafka,分布式

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

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

Producer发送ProducerRecord的流程:

主线程将ProducerRecord加工处理后发送至RecordAccumulator尾部
RecordAccumulator根据ProducerRecord分区选择对应的ProducerBatch
RecordAccumulator根据内存复用原则和ProducerBatch大小决定是否新建
Sender线程从RecordAccumulator头部获取ProducerBatch
将<分区, <Deque<ProducerBatch>>形式变为<Node, List<ProducerBatch>>
再根据各种协议请求转换为<Node, Request>形式
发送前以Map<nodeId, Deque<Request>>缓存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 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[])

Broker仅能接受字节数组形式的数据(接收后会对其反序列化)
Producer使用的Serializer需和Consumer使用的反序列化器需对应
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分区的默认规则

ProducerRecord中指定partition字段, 则略过Partitioner
Partitioner的分区计算受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(Map<String, ?> 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(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId)

// 提交事务
void commitTransaction();

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