Kafka核心API生产者实战详解

DBC 2.4K 0

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上去的，这样性能才可能题高
  • 生产者常见配置
温馨提示

#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

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();


    }
控制台输出

3.ProducerRecord介绍和key的作用

ProducerRecord（简称PR）

  • 发送给Kafka Broker的key/value 值对, 封装基础数据信息
温馨提示

-- Topic （名字）
-- PartitionID (可选)
-- Key(可选)
-- Value

Kafka核心API生产者实战详解插图
Kafka核心API生产者实战详解插图2

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个副本

Kafka核心API生产者实战详解插图4
Kafka核心API生产者实战详解插图6

第二步：发送消息

    /**
     * 发送消息携带回调函数，指定某个分区
     *
     * 实现顺序消息
     */
    @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分区规则实战

源码解读默认分区器

org.apache.kafka.clients.producer.internals.DefaultPartitioner

自定义分区规则

  • 创建类，实现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，转载请注明。

发表评论 取消回复
表情 图片 链接 代码

分享