美文网首页
Kafka生产者

Kafka生产者

作者: 橙皇cc | 来源:发表于2020-09-26 21:05 被阅读0次

    Kafka生产者概览

    在这里插入图片描述
    流程概述: 创建一个ProducerRecord对象,需要包含topic和要发送的内容,还可以指定键或分区。在发送ProducerRecord对象时,生产者要先把键和值对象序列化成字节数组。然后数据传给分区器,如果对象中指定了分区,那分区器直接把指定的分区返回,如果没有指定,分区器会根据ProducerRecord的键选择一个分区。这条记录随后被添加到记录批次里,这个批次里的所有消息会被发送到相同的主题和分区上。最后服务器收到消息成功写入会返回一个RecordMetaData对象,包含了主题和分区信息,以及记录在分区里的偏移量。如果写入失败,就会返回错误信息,生产者收到错误后会尝试重新发送,几次之后如果还是失败就返回错误信息。

    创建Kafka生产者

    这里只指定了必要的属性,其他使用默认配置

    private Properties kafkaProps = new Properties();
    kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
    kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps);
    
    1. 指定了broker的地址清单,建议至少提供两个broker信息,其中一个宕机,生产者仍然能够连接到集群上。
    2. 指定了消息的键和值都是String

    Kafka客户端默认提供了ByteArraySerializer、StringSerializer和IntegerSerializer接口的类。

    发送消息到Kafka

    简单配置

    ProducerRecord<String, String> record = new ProducerRecord<>("topic11", "key111", "value111");
    try {
        producer.send(record);
    } catch (Exception e) {
        e.printStackTrace();
    }
    

    这种方式不关心是否正常送达,虽然生产者会自动尝试重发,但不能完全保证消息不丢失。

    同步发送消息

    ProducerRecord<String, String> record = new ProducerRecord<>("topic11", "key111", "value111");
    try {
        producer.send(record).get();
    } catch (Exception e) {
        e.printStackTrace();
    }
    

    send()方法会返回一个Future对象,调用get()方法进行等待可以知道消息是否发送成功。

    异步发送消息

    public class DemoProducerCallback implements Callback {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception != null){
                exception.printStackTrace();
            }
        }
    }
    
    ProducerRecord<String, String> record = new ProducerRecord<>("topic11", "key111", "value111");
    producer.send(record ,new DemoProducerCallback()); 
    

    异步发送消息不等待响应,会提高发送速度,但如果遇到发送失败,可以使用回调进行一些处理。

    顺序保证

    Kafka可以保证同一个分区里的消息是有序的。如果把retries设为非零整数,把max.in.flight.requests.per.connection设为比1大的数。如果第一个批次消息写入失败,而第二个批次写入成功,broker会重试写入第一个批次,如果此时第一个批次也写入成功,那么两个批次的顺序就反过来了。如果某些场景要求消息是有序的,可以把max.in.flight.requests.per.connection设为1,这样在生产者尝试发送第一批消息时,不会有其他的消息发送给broker。不过这样做严重影响生产者的吞吐量。

    分区

    Kafka的消息是一个个键值对,ProducerRecord对象可以 只包含目标主题和值,键可以设置为默认的null。键有两个用途:可以作为消息的附加信息,也可以用来决定消息该被写到topic的哪个分区。拥有相同键的消息将被写到同一个分区。

    包含键值创建:

    ProducerRecord<String, String> record = new ProducerRecord<>("topic11", "key111", "value111");
    

    不包含键值创建:

    ProducerRecord<String, String> record = new ProducerRecord<>("topic11",  "value111");
    

    不包含键值的话,使用默认的分区器,记录将被随机地发送到topic各个可用的分区上。如果键不为空,使用了默认的分区器,那么Kafka会对键进行散列,然后根据散列值把消息映射到特定的分区上,同一个键总是被映射到同一个分区上,这也意味着,有可能会写入到不可用的分区而发生错误。
    只有在不改变topic分区数量的情况下,键与分区的映射保持不变,一旦topic增加了新的分区,就无法保证了。下面实现自定义分区策略解决这一问题。

    自定义分区策略

    如下对target键做专一分区,其他键采用默认随机分配。

    package com.qax.ngsoc.workorder.message.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.record.InvalidRecordException;
    import org.apache.kafka.common.utils.Utils;
    
    import java.util.List;
    import java.util.Map;
    
    public class MyPartitioner implements Partitioner {
    
        @Override
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            int size = partitions.size();
    
            if (keyBytes == null || (!(key instanceof String))) {
                throw new InvalidRecordException("只接收字符串作为键");
            }
    
            if ("target".equals(key)) {
                return size;
            }
    
            return (Math.abs(Utils.murmur2(keyBytes)) % (size - 1));
        }
    
        @Override
        public void close() {
    
        }
    
        @Override
        public void configure(Map<String, ?> configs) {
    
        }
    }
    

    相关文章

      网友评论

          本文标题:Kafka生产者

          本文链接:https://www.haomeiwen.com/subject/epshuktx.html