一、生产者发送到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上去的，这样性能才可能题高
三、生产者常见配置
官方文档 http://kafka.apache.org/documentation/#producerconfigs
#kafka地址,即broker地址 bootstrap.servers #当producer向leader发送数据时，可以通过request.required.acks参数来设置数据可靠性的级别,分别是0, 1，all。 acks #请求失败，生产者会自动重试，指定是0次，如果启用重试，则会有重复消息的可能性 retries #每个分区未发送消息总字节大小,单位：字节，超过设置的值就会提交数据到服务端，默认值是16KB batch.size # 默认值就是0，消息是立刻发送的，即便batch.size缓冲空间还没有满，如果想减少请求的数量，可以设置 linger.ms 大于#0，即消息在缓冲区保留的时间，超过设置的值就会被提交到服务端 # 通俗解释是，本该早就发出去的消息被迫至少等待了linger.ms时间，相对于这时间内积累了更多消息，批量发送 减少请求 #如果batch被填满或者linger.ms达到上限，满足其中一个就会被发送 linger.ms # buffer.memory的用来约束Kafka Producer能够使用的内存缓冲的大小的，默认值32MB。 # 如果buffer.memory设置的太小，可能导致消息快速的写入内存缓冲里，但Sender线程来不及把消息发送到Kafka服务器 # 会造成内存缓冲很快就被写满，而一旦被写满，就会阻塞用户线程，不让继续往Kafka写消息了 # buffer.memory要大于batch.size，否则会报申请内存不足的错误，不要超过物理内存，根据实际情况调整 buffer.memory # key的序列化器，将用户提供的 key和value对象ProducerRecord 进行序列化处理，key.serializer必须被设置，即使 #消息中没有指定key，序列化器必须是一个实现org.apache.kafka.common.serialization.Serializer接口的类，将#key序列化成字节数组。 key.serializer value.serializer
四、代码封装配置属性
public static Properties getProperties(){
Properties props = new Properties();
props.put("bootstrap.servers", "xxx.xxx.xxx.xx:9092");
//props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "xxx.xxx.xxx.x: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);
props.put("linger.ms", 5);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
return props;
}
发送消息
@Test
public void testSend(){
Properties properties = getProperties();
Producer producer = new KafkaProducer<>(properties);
for(int i=0;i<3 ;i++){
Future future = producer.send(new ProducerRecord<>(TOPIC_NAME,"wnn-key"+i, "wnn-content-value"+i));
try {
//不关心结果则不用写这些内容
Recordmetadata recordmetadata = future.get();
// topic - 分区编号@offset
System.err.println("发送状态："+recordmetadata.toString());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
producer.close();
}
五、第四步骤中发送消息使用到了ProducerRecord（简称PR）
PR主要用来 发送给Kafka Broker的key/value 值对, 封装基础数据信息Topic(名字)PartitionID(可选) Key(可选)-- Value
key默认是null，大多数应用程序会用到key
如果key为空，kafka使用默认的partitioner，使用RoundRobin算法将消息均衡地分布在各个partition上
如果key不为空，kafka使用自己实现的hash方法对key进行散列，决定消息该被写到Topic的哪个partition，拥有相同key的消息会被写到同一个partition，实现顺序消息
六、生产者发送消息是异步调用，怎么知道是否有异常
发送消息配置回调函数即可， 该回调方法会在 Producer 收到 ack 时被调用，为异步调用
回调函数有两个参数 Recordmetadata 和 Exception，如果 Exception 是 null，则消息发送成功，否则失败
@Test
public void testSendWithCallback(){
Properties properties = getProperties();
Producer producer = new KafkaProducer<>(properties);
for(int i=0;i<3 ;i++) {
producer.send(new ProducerRecord<>(TOPIC_NAME, "wnn-key" + i, "wnn-content-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();
}
七、producer生产者发送指定分区
创建topic，配置5个分区，1个副本
@Test
public void testSendWithCallbackAndPartition(){
Properties properties = getProperties();
Producer producer = new KafkaProducer<>(properties);
for(int i=0;i<10 ;i++) {
producer.send(new ProducerRecord<>("wnn-topic-test-12.18-fenqu", 4,"wnn-key" + i, "wnn-content-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();
}
八、Kafka 生产者自定义partition分区规则
源码解读默认分区器
org.apache.kafka.clients.producer.internals.DefaultPartitioner
更改后的 可以根据业务需求获取到对应参数后进行更改
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 WnnPartitioner 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("wnn".equals(key)){
return 0;
}
List 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 configs) {
}
}
调用代码
@Test
public void testSendWithPartitionStrategy(){
Properties properties = getProperties();
properties.put("partitioner.class", "net.wnn.wnnkafka.config.WnnPartitioner");
Producer producer = new KafkaProducer<>(properties);
for(int i=0;i<10 ;i++) {
producer.send(new ProducerRecord<>("wnn-topic-test-12.18-fenqu", null, "wnn-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();
}
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）