1.Kafka的producer生产者发送到Broker分区策略讲解
- 生产者发送到broker里面的流程是怎样的呢,一个 topic 有多个 partition分区,每个分区又有多个副本
- 如果指定Partition ID,则PR被发送至指定Partition (ProducerRecord)
- 如果未指定Partition ID,但指定了Key, PR会按照hash(key)发送至对应Partition
- 如果未指定Partition ID也没指定Key,PR会按照默认 round-robin轮训模式发送到每个Partition
- 消费者消费partition分区默认是range模式
- 如果同时指定了Partition ID和Key, PR只会发送到指定的Partition (Key不起作用,代码逻辑决定)
- 注意:Partition有多个副本,但只有一个replicationLeader负责该Partition和生产者消费者交互
- 生产者到broker发送流程
- Kafka的客户端发送数据到服务器,不是来一条就发一条,会经过内存缓冲区(默认是16KB),通过KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里,然后把很多消息收集到的Batch里面,再一次性发送到Broker上去的,这样性能才可能题高
- 生产者常见配置
2.Kafka核心API模块-producer API讲解实战
封装配置属性
package net.xdclass.xdclasskafka; import org.apache.kafka.clients.producer.*; import org.apache.kafka.clients.producer.internals.FutureRecordMetadata; import org.junit.jupiter.api.Test; import java.util.Properties; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; public class KafkaProducerTest { private static final String TOPIC_NAME = "xdclass-sp-topic-test"; public static Properties getProperties(){ Properties props = new Properties(); props.put("bootstrap.servers", "112.74.55.160:9092"); //props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "112.74.55.160:9092"); // 当producer向leader发送数据时,可以通过request.required.acks参数来设置数据可靠性的级别,分别是0, 1,all。 props.put("acks", "all"); //props.put(ProducerConfig.ACKS_CONFIG, "all"); // 请求失败,生产者会自动重试,指定是0次,如果启用重试,则会有重复消息的可能性 props.put("retries", 0); //props.put(ProducerConfig.RETRIES_CONFIG, 0); // 生产者缓存每个分区未发送的消息,缓存的大小是通过 batch.size 配置指定的,默认值是16KB props.put("batch.size", 16384); /** * 默认值就是0,消息是立刻发送的,即便batch.size缓冲空间还没有满 * 如果想减少请求的数量,可以设置 linger.ms 大于0,即消息在缓冲区保留的时间,超过设置的值就会被提交到服务端 * 通俗解释是,本该早就发出去的消息被迫至少等待了linger.ms时间,相对于这时间内积累了更多消息,批量发送减少请求 * 如果batch被填满或者linger.ms达到上限,满足其中一个就会被发送 */ props.put("linger.ms", 5); /** * buffer.memory的用来约束Kafka Producer能够使用的内存缓冲的大小的,默认值32MB。 * 如果buffer.memory设置的太小,可能导致消息快速的写入内存缓冲里,但Sender线程来不及把消息发送到Kafka服务器 * 会造成内存缓冲很快就被写满,而一旦被写满,就会阻塞用户线程,不让继续往Kafka写消息了 * buffer.memory要大于batch.size,否则会报申请内存不#足的错误,不要超过物理内存,根据实际情况调整 * 需要结合实际业务情况压测进行配置 */ props.put("buffer.memory", 33554432); /** * key的序列化器,将用户提供的 key和value对象ProducerRecord 进行序列化处理,key.serializer必须被设置, * 即使消息中没有指定key,序列化器必须是一个实 org.apache.kafka.common.serialization.Serializer接口的类, * 将key序列化成字节数组。 */ props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer"); return props; } }
生产者投递消息API实战(同步发送)
/** * send()方法是异步的,添加消息到缓冲区等待发送,并立即返回 * 生产者将单个的消息批量在一起发送来提高效率,即 batch.size和linger.ms结合 * * 实现同步发送:一条消息发送之后,会阻塞当前线程,直至返回 ack * 发送消息后返回的一个 Future 对象,调用get即可 * * 消息发送主要是两个线程:一个是Main用户主线程,一个是Sender线程 * 1)main线程发送消息到RecordAccumulator即返回 * 2)sender线程从RecordAccumulator拉取信息发送到broker * 3) batch.size和linger.ms两个参数可以影响 sender 线程发送次数 * * */ @Test public void testSend(){ Properties properties = getProperties(); Producer<String,String> producer = new KafkaProducer<>(properties); for(int i=0;i<3 ;i++){ Future<RecordMetadata> future = producer.send(new ProducerRecord<>(TOPIC_NAME,"xdclass-key"+i, "xdclass-value"+i)); try { //不关心结果则不用写这些内容 RecordMetadata recordMetadata = future.get(); // topic - 分区编号@offset System.out.println("发送状态:"+recordMetadata.toString()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } producer.close(); }
控制台输出
发送状态:xdclass-sp-topic-test-0@0
发送状态:xdclass-sp-topic-test-0@1
发送状态:xdclass-sp-topic-test-1@0
3.ProducerRecord介绍和key的作用
ProducerRecord(简称PR)
- 发送给Kafka Broker的key/value 值对, 封装基础数据信息
key默认是null,大多数应用程序会用到key
- 如果key为空,kafka使用默认的partitioner,使用RoundRobin算法将消息均衡地分布在各个partition上
- 如果key不为空,kafka使用自己实现的hash方法对key进行散列,决定消息该被写到Topic的哪个partition,拥有相同key的消息会被写到同一个partition,实现顺序消息
4.Kafka核心API模块-producerAPI回调函数实战
生产者发送消息是异步调用,怎么知道是否有异常?
- 发送消息配置回调函数即可, 该回调方法会在 Producer 收到 ack 时被调用,为异步调用
- 回调函数有两个参数 RecordMetadata 和 Exception,如果 Exception 是 null,则消息发送成功,否则失败
异步发送配置回调函数
/** * 发送消息携带回调函数 */ @Test public void testSendWithCallback(){ Properties properties = getProperties(); Producer<String,String> producer = new KafkaProducer<>(properties); for(int i=0;i<3 ;i++) { producer.send(new ProducerRecord<>(TOPIC_NAME, "xdclass-key" + i, "xdclass-value" + i), new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if(exception == null){ System.err.println("发送状态:"+metadata.toString()); } else { exception.printStackTrace(); } } }); } producer.close(); }
5.producer生产者发送指定分区实战
第一步:创建topic,配置5个分区,1个副本
第二步:发送消息
/** * 发送消息携带回调函数,指定某个分区 * * 实现顺序消息 */ @Test public void testSendWithCallbackAndPartition(){ Properties properties = getProperties(); Producer<String,String> producer = new KafkaProducer<>(properties); for(int i=0;i<10 ;i++) { producer.send(new ProducerRecord<>("xdclass-v1-sp-topic-test", 4,"xdclass-key" + i, "xdclass-value" + i), new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if(exception == null){ System.err.println("发送状态:"+metadata.toString()); } else { exception.printStackTrace(); } } }); } producer.close(); }
6.Kafka 生产者自定义partition分区规则实战
自定义分区规则
- 创建类,实现Partitioner接口,重写方法
- 配置 partitioner.class 指定类即可
package net.xdclass.xdclasskafka.config; import org.apache.kafka.clients.producer.Partitioner; import org.apache.kafka.common.Cluster; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.utils.Utils; import java.util.List; import java.util.Map; public class XdclassPartitioner implements Partitioner { @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { if (keyBytes == null) { throw new IllegalArgumentException("key 参数不能为空"); } if("xdclass".equals(key)){ return 0; } List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } @Override public void close() { } @Override public void configure(Map<String, ?> configs) { } }
测试发送
/** * 自定义分区策略 */ @Test public void testSendWithPartitionStrategy() { Properties properties = getProperties(); properties.put("partitioner.class", "net.xdclass.xdclasskafka.config.XdclassPartitioner"); Producer<String, String> producer = new KafkaProducer<>(properties); for (int i = 0; i < 10; i++) { producer.send(new ProducerRecord<>("xdclass-v1-sp-topic-test", "xdclass", "xdclass-value" + i), new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { System.err.println("发送状态:" + metadata.toString()); } else { exception.printStackTrace(); } } }); } producer.close(); }
本文作者为DBC,转载请注明。