Kafka自定义分区器/提高生产吞吐量/数据可靠性

目录

自定义分区器

生产经验--- 生产者如何提高吞吐量

 生产经验--数据可靠性

自定义分区器

1、创建一个新的包：Partitioner

创建一个分区器

topic                   主题

key                     消息的 key

keyBytes            消息的 key 序列化后的字节数组

value                  消息的 value

valueBytes         消息的 value 序列化后的字节数组

cluster                集群元数据可以查看分区信息

package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

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) {

     //获取数据atguigu  hello
        String msgValues = value.toString();

        int partition=0;

        if(msgValues.contains("atguigu")) {
            partition=0;
        }else{
            partition=1;
        }


        return partition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map map) {

    }
}

关联自定义分区器

//关联自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.atguigu.kafka.producer.MyPartitioner");

 测试自定义分区器

测试1

将发送消息设置为atguigu；查看它的分区

测试结果如图所示：

测试2

将发送消息设置为hello；查看它的分区

测试结果如图所示： 

 

生产经验--- 生产者如何提高吞吐量

batch.size批次大小

linger.ms等待时间

compression.type压缩

RecordAccumulator缓冲区大小

以上参数都可以影响到生产者的吞吐量；

batch.size:批次大小，默认16k

linger.ms：等待时间，修改为5-100ms

compression.type：压缩snappy

RecordAccumulator：缓冲区大小，修改为64m

 创建一个包：CustomProducerParameters

package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;


public class CustomProducerParameters {

    public static void main(String[] args)  {

        //配置信息
        Properties properties = new Properties();

        //连接Kafka集群

        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");

        //序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        //缓冲区大小
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);

        //批次大小
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);

        //linger.ms
        properties.put(ProducerConfig.LINGER_MS_CONFIG,1);

        //压缩
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");

        //1、创建生产者
        KafkaProducer kafkaProducer = new KafkaProducer<>(properties);

        //2、发送数据
        for (int i=0; i<5; i++){
            kafkaProducer.send(new ProducerRecord<>("first","hello"+i));
        }

        //3、关闭资源
        kafkaProducer.close();

    }

或者可以Ctrl+c ；Ctrl+v；复制CustomProducer包；修改名称为CustomProducerParameters

然后在包下添加如下代码即可

        //缓冲区大小
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);

        //批次大小
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);

        //linger.ms
        properties.put(ProducerConfig.LINGER_MS_CONFIG,1);

        //压缩
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");

测试

结果运行如下：

 生产经验--数据可靠性

发送流程图：

ACK应答级别 

 

以上三张图片来源于尚硅谷 

 可靠性总结：

acks=0，生产者发送过来数据就不管了，可靠性差，效率搞；

acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；

acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；

生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

数据重复分析

下图来源于尚硅谷

 代码配置

package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerAcks {

    public static void main(String[] args) {

        // 0配置
        Properties properties = new Properties();

        // 连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");

        // 指定对应的key和value的序列化类型
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //acks
        properties.put(ProducerConfig.ACKS_CONFIG,"1");

        //重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG,"3");

        // 1创建kafka生产者对象

        // “”hello
        KafkaProducer kafkaProducer = new KafkaProducer<>(properties);


        // 2发送数据
        for (int i=0; i<5; i++) {

            kafkaProducer.send(new ProducerRecord<>("first","HYF"+i));

        }
        // 3关闭资源
        kafkaProducer.close();
    }



}

只需要在之前的包中插入以下两行代码即可
//acks
properties.put(ProducerConfig.ACKS_CONFIG,"1");

//重试次数
properties.put(ProducerConfig.RETRIES_CONFIG,"3");

测试

测试结果如图