一、消息分区机制producer发送过程有个很重要的步骤就是确定发送的消息在哪个topic分区中。Producer提供了分区策略和对应的分区器partitioner供用户使用。新版本的会把相同key的消息发送到partition上如果没有指定key则会通过轮询分配均匀在topic所在分区而对于旧版本的无法分配均匀。自定义分区机制对于有key的消息java版本的producer会通过自己的算法计算key的哈希值然后在总分区取模分配到目标分区。但有的时候用户想实现自己的分区策略而这又是默认partitioner无法实现的那么此刻就可以用producer提供的自定义分区策略。代码语言javascriptAI代码解释/** * author keying */ public class AuditPartitioner implements Partitioner { private Random random; Override public int partition(String topic, Object keyObj, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { String key (String) keyObj; ListPartitionInfo partitionInfoList cluster.availablePartitionsForTopic(topic); int auditPartition partitionInfoList.size() - 1; return key null || key.isEmpty() || !key.contains(audit) ? random.nextInt(partitionInfoList.size() - 1) : auditPartition; //return 0; } Override public void close() { } Override public void configure(MapString, ? configs) { random new Random(); } }若自定义分区机制则需要做两件事1、先定义一个类实现org.apache.kafka.clients.producer.Partitioner接口主要重写partition方法。2、在构造kafkaProducer的时候propertites设置partitioner参数。Partition方法里主要接受参数有topickey和value还有集群元数据信息一起来确定目标分区而close方法则是用于关闭partitioner的主要是为了关闭那些创建partitioner时初始化的系统资源等。举个例子如何实现自定义的partitioner呢假设我们有个类似审计功能审计功能发送kafka的时候可以给他分配字符串“audit”我们想让这类消息发到topic最后一个分区上便于后续统一处理而对于相同topic下的其他消息则采用随机发送的策略发送到其他分区上。所以用户可以根据key来指定一些策略还可以根据value信息做一些定制化分区策略。二、消息序列化网络中发送数据都是以字节的方式kafka也不例外它可以是字符串一个整数一个数组或者其他任意对象类型。序列化器serializer负责在producer发送将消息转换成字节数组而与之相反解序列化器deserializer则用于将consumer接受到的字节数组转换成相应的对象。Kafka1.0.0默认提供十几种序列化器常见的serializer用的是StringSerializer然后其他的还有LongSerializerIntegerSerializer等。如果是复杂的类型比如Avro则需要自定义序列化。三、Producer拦截器Producer拦截器相当于一个新的功能他可以在producer发送消息之后以及回调之前有机会对消息做些定制化需求比如修改消息等。同时producer允许用户指定多个interceptor按序作用于同一条消息从而形成一个拦截器intercetpor的实现接口是producerInterceptor其定义方法如下onSend(producerRecord)该方法封装进kafkaProducer.send方法中即他运行在用户主线程中。Producer确保在消息被序列化以计算分区前调用该方法。用户可以在该方法对消息做任何处理但最好不要修改消息的所属topic和分区否则影响分区计算。onAcknowledgementrecordMetadataException该消息会在被应答之前或者消息发送失败时候调用并且通常在producer回调触发之前调用。OnAcknoewledgement运行在producer的I/O线程中因此不要在该方法放入很重的逻辑否则会拖慢producer的消息发送效率。Close:关闭interceptor主要做一些资源清理工作。如前所述interceptor可能运行在多个线程中因此具体实现时候需要用户自行确认保护线程安全。若指定多个interceptor则producer将按照指定顺序调用他们同时把每个interceptor中捕获的异常记录到错误日志中而不是向上传递。代码语言javascriptAI代码解释/** * author keying * date 2022-08-07 17:24:21 */ public class OneInterceptor implements ProducerInterceptorString, String { Override public ProducerRecordString, String onSend(ProducerRecord record) { return new ProducerRecord(record.topic(), record.partition(), record.timestamp(), System.currentTimeMillis() , record.value().toString()); } Override public void onAcknowledgement(RecordMetadata metadata, Exception exception) { } Override public void close() { } Override public void configure(MapString, ? configs) { } } /** * author keying * date 2022-08-07 17:27:40 */ public class TwoInterceptor implements ProducerInterceptorString, String { private int errorCounter 0; private int successCounter 0; Override public ProducerRecordString, String onSend(ProducerRecordString, String record) { return null; } Override public void onAcknowledgement(RecordMetadata metadata, Exception exception) { if (exception null) { successCounter; } else { errorCounter; } } Override public void close() { } Override public void configure(MapString, ? configs) { System.out.println(成功:successCounter); System.out.println(失败errorCounter); } }上面例子是实现一个简单的双inteceptor组成的拦截器第一个拦截器会在消息发送前将时间戳加入到value第二个拦截器则会统计成功和失败的次数。四、无消息丢失配置Producer采用的是异步发送消息机制kafkaProducer.send方法仅仅把消息放入缓冲区由一个专属的I/O线程负责提取缓冲区的消息并封装到batch中然后发送出去。显然整个过程存在数据丢失的窗口若I/O线程在发送之前崩溃则数据会丢失。另一个问题则是消息会乱序比如客户端依次发送两条消息到不同的分区Producer.send(records1);和producer.send(records2)若此刻某些原因网络出现瞬时抖动导致records1发送失败同时kafka又配置了重试机制max.in.flight.requests.per.connection大于1默认是5这样会造成消息乱序而实际场景很多情况需要包装按顺序消费。所以这两个问题kafka该如何规避呢首先消息丢失很容易想到kafka的同步发送但这样性能会很差并不在实际场景中推荐使用。如何配置保证消息不会丢失呢Block.on.buffer.full trueAcksall 或者 -1RetriesInteger.MAX_VALUEMax.in.flight.request.per.connection1使用回调机制的send发送消息CallBack逻辑中显式立即关闭producer使用close(0)Unclean.leader.election.enablefalseReplication.factor3Min.insync.replicas 2Replication.factormin.insync.replicasEnable.auto.commitfalseProducer端配置Block.on.buffer.full true实际上这个参数在kafka0.9.0版本已经被标记为deprecated的并且使用max.block.ms替代但还是推荐用户显示的设置它为true使得内存缓冲区被填满时producer处于阻塞状态并且停止接受新消息而不是抛出异常。否则producer生产速度过快会耗尽缓冲区新版本0.10.0.0不用管这个参数直接设置max.block.ms参数。Acks all很好理解就是所有leader broker和副本replict里的follower都收到消息才回复producer消息成功发送。RetriesInteger.MAX_VALUE这里设置无限大有点极端想表达的是无线重试但放心这里不会重试那些无法恢复的错误只会重试那些可恢复的异常所以可以放心的设置比较大的值保证消息不会丢失。max.in.flight.request.per.connection1设置为1防止消息在topic下乱序这个设置的效果限制了producer在单个broker上连续发送的未响应请求数量。因此如果设置成1则producer在某个broker发送响应之前将无法再给broker发送producer请求。使用带回调的send普通的send官方解释是fire and forget只管把消息发出去不管后续如果发送失败不会收到任何通知这里肯定要带回调的send发送。CallbackBack逻辑中显式处理立刻关闭producer在calllback失败处逻辑立即使用kafkaProcuer.close(0)这样做的目的就是为了防止消息乱序问题。若不使用close关闭默认情况下producer会被允许将未完成的消息发送出去这样可能造成消息乱序。Broker端配置Unclean.leader.election.eable false关闭unclean leader选举即不允许非isr中的副本被选举成leader从而避免broker端因为日志水位截断造成数据丢失。Replication.factor3 :设置成3主要参考业界的三备份原则强调多个副本才好。Min.insync.replias1用于控制某条消息至少被写入ISR中多个副本才算成功大于1代表提升持久性只有在acks设置成-1或者all的时候才生效。确保 replication.factorsmin.insync.replicas 若两者相等则只要有一个副本挂掉则分区无法正常使用虽然持久性很高但可用性被降低建议 replication.factory min.insync.replicas 1。