Kafka Java客户端数据生产流程解析，从发送类型实现代码到

作者: 架构小菜 | 来源:发表于2021-01-15 10:27 被阅读0次

kafka-第二章-生产者
Kafka Java客户端数据生产流程解析，从发送类型实现代码到
Kafka高级特性解析(一)
哈哈
Spark Structured Streaming 解析 JS
Kafka 生产者调优
Kafka学习——生产者和消费者客户端
Python requests 爬取淘宝商品数据，并连接数据库，
0005.ObjectMapper解析JSON(数组)字符串
2网络协议基础

一、消息发送

1.Kafka Java客户端数据生产流程解析

首先要构造一个 ProducerRecord 对象，该对象可以声明主题Topic、分区Partition、键 Key以及值 Value，主题和值是必须要声明的，分区和键可以不用指定。
调用send() 方法进行消息发送。
因为消息要到网络上进行传输，所以必须进行序列化，序列化器的作用就是把消息的 key 和 value对象序列化成字节数组。
接下来数据传到分区器，如果之间的 ProducerRecord 对象指定了分区，那么分区器将不再做任何事，直接把指定的分区返回；如果没有，那么分区器会根据 Key 来选择一个分区，选择好分区之后，生产者就知道该往哪个主题和分区发送记录了。
接着这条记录会被添加到一个记录批次里面，这个批次里所有的消息会被发送到相同的主题和分区。会有一个独立的线程来把这些记录批次发送到相应的 Broker 上。
Broker成功接收到消息，表示发送成功，返回消息的元数据（包括主题和分区信息以及记录在分区里的偏移量）。发送失败，可以选择重试或者直接抛出异常。

依赖的包 <kafka.version>2.0.0</kafka.version>

<dependency>
    <groupId>org.apache.kafka</groupId> 
    <artifactId>kafka_${scala.version}</artifactId> 
    <version>${kafka.version}</version> 
    <exclusions> 
        <exclusion> 
            <groupId>org.apache.zookeeper</groupId> 
            <artifactId>zookeeper</artifactId> 
        </exclusion> 
        <exclusion> 
            <groupId>org.slf4j</groupId> 
            <artifactId>slf4j-log4j12</artifactId> 
        </exclusion> 
        <exclusion> 
            <groupId>log4j</groupId> 
            <artifactId>log4j</artifactId> 
        </exclusion> 
    </exclusions> 
</dependency>

2.必要参数配置

见代码库：com.heima.kafka.chapter2.KafkaProducerAnalysis

public static Properties initConfig() { 
    Properties props = new Properties(); 
    // 该属性指定 brokers 的地址清单，格式为 host:port。清单里不需要包含所有的 broker 地址， 
    // 生产者会从给定的 broker 里查找到其它 broker 的信息。——建议至少提供两个 broker 的信息，因为一旦其中一个宕机，生产者仍然能够连接到集群上。 
    props.put("bootstrap.servers", brokerList); 
    // 将 key 转换为字节数组的配置，必须设定为一个实现了 org.apache.kafka.common.serialization.Serializer 接口的类， 
    // 生产者会用这个类把键对象序列化为字节数组。 
    // ——kafka 默认提供了 StringSerializer和 IntegerSerializer、 ByteArraySerializer。当然也可以自定义序列化器。 
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
    // 和 key.serializer 一样，用于 value 的序列化 
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    // 用来设定KafkaProducer对应的客户端ID，默认为空，如果不设置KafkaProducer会自动 生成一个非空字符串。 
    // 内容形式如："producer-1" 
    props.put("client.id", "producer.client.id.demo"); 
    return props; 
}

Properties props = initConfig(); 
    KafkaProducer<String, String> producer = new KafkaProducer<>(props); 
//  KafkaProducer<String, String> producer = new KafkaProducer<>(props, 
// new StringSerializer(), new StringSerializer()); 
//生成 ProducerRecord 对象，并制定 Topic，key 以及 value 
    ProducerRecord<String, String> record = new ProducerRecord<>(topic, "hello, Kafka!"); 
    try {
        // 发送消息 
        producer.send(record);
    }

3.发送类型

发送即忘记

producer.send(record)

同步发送

//通过send()发送完消息后返回一个Future对象,然后调用Future对象的get()方法等待kafka响应 
//如果kafka正常响应，返回一个RecordMetadata对象，该对象存储消息的偏移量 
// 如果kafka发生错误，无法正常响应，就会抛出异常，我们便可以进行异常处理 
producer.send(record).get();

异步发送

producer.send(record, new Callback() { 
    public void onCompletion(RecordMetadata metadata, Exception exception) { 
    if (exception == null) { 
        System.out.println(metadata.partition() + ":" + metadata.offset()); 
    } 
} 
});

4.序列化器

消息要到网络上进行传输，必须进行序列化，而序列化器的作用就是如此。

Kafka 提供了默认的字符串序列化器（org.apache.kafka.common.serialization.StringSerializer），还有整型（IntegerSerializer）和字节数组（BytesSerializer）序列化器，这些序列化器都实现了接口（org.apache.kafka.common.serialization.Serializer）基本上能够满足大部分场景的需求。

5.自定义序列化器

见代码库：com.heima.kafka.chapter2.CompanySerializer

/**
    * 自定义序列化器
    */
public class CompanySerializer implements Serializer<Company> { 
    @Override 
    public void configure(Map configs, boolean isKey) { 
    }
    
    @Override 
    public byte[] serialize(String topic, Company data) { 
        if (data == null) { 
            return null; 
        }
        byte[] name, address; 
        try {
            if (data.getName() != null) { 
                name = data.getName().getBytes("UTF-8"); 
            } else { 
                name = new byte[0]; 
            }
            if (data.getAddress() != null) { 
                address = data.getAddress().getBytes("UTF-8"); 
            } else { 
                address = new byte[0]; 
            }
            ByteBuffer buffer = ByteBuffer. allocate(4 + 4 + name.length + address.length); 
            buffer.putInt(name.length); 
            buffer.put(name); 
            buffer.putInt(address.length); 
            buffer.put(address); 
            return buffer.array(); 
        } catch (UnsupportedEncodingException e) { 
            e.printStackTrace(); 
        }
        return new byte[0]; 
    }
    
    @Override 
    public void close() { 
    } 
}

使用自定义的序列化器

见代码库：com.heima.kafka.chapter2.ProducerDefineSerializer

public class ProducerDefineSerializer { 
    public static final String brokerList = "localhost:9092"; 
    public static final String topic = "heima"; 
    
    public static void main(String[] args) throws ExecutionException, InterruptedException { 
        Properties properties = new Properties(); 
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); 
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CompanySerializer.class.getName()); 
// properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
// ProtostuffSerializer.class.getName()); 
    properties.put("bootstrap.servers", brokerList); 
    
    KafkaProducer<String, Company> producer = new KafkaProducer<>(properties);
    Company company = Company.builder().name("kafka") .address("北京").build(); 
// Company company = Company.builder().name("hiddenkafka") 
// .address("China").telphone("13000000000").build(); 
    ProducerRecord<String, Company> record = new ProducerRecord<>(topic, company); 
    producer.send(record).get(); 
    } 
}

6.分区器

本身kafka有自己的分区策略的，如果未指定，就会使用默认的分区策略：

Kafka根据传递消息的key来进行分区的分配，即hash(key) % numPartitions。如果Key相同的话，那么就会分配到统一分区。

源代码org.apache.kafka.clients.producer.internals.DefaultPartitioner分析

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { 
    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); 
    int numPartitions = partitions.size(); 
    if (keyBytes == null) { 
        int nextValue = this.nextValue(topic); 
        List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); 
        if (availablePartitions.size() > 0) { 
            int part = Utils.toPositive(nextValue) % availablePartitions.size(); 
            return ((PartitionInfo)availablePartitions.get(part)).partition(); 
        } else { 
            return Utils.toPositive(nextValue) % numPartitions; 
        } 
    } else { 
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; 
    } 
}

自定义分区器见代码库 com.heima.kafka.chapter2.DefinePartitioner

/**
    * 自定义分区器 
    */ 
public class DefinePartitioner implements Partitioner { 
    private final AtomicInteger counter = new AtomicInteger(0); 
    
    @Override 
    public int partition(String topic, Object key, byte[] keyBytes,Object value, byte[] valueBytes, Cluster cluster) { 
    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); 
    int numPartitions = partitions.size(); 
    if (null == keyBytes) { 
        return counter.getAndIncrement() % numPartitions; 
        } else return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; 
    }
    @Override 
    public void close() { 
    }
    @Override 
    public void configure(Map<String, ?> configs) { 
    } 
}

实现自定义分区器需要通过配置参数ProducerConfig.PARTITIONER_CLASS_CONFIG来实现

// 自定义分区器的使用 
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,DefinePartitioner.class.getNam e());

7.拦截器

Producer拦截器(interceptor)是个相当新的功能，它和consumer端interceptor是在Kafka 0.10版本被引入的，主要用于实现clients端的定制化控制逻辑。

生产者拦截器可以用在消息发送前做一些准备工作。

使用场景

按照某个规则过滤掉不符合要求的消息
修改消息的内容
统计类需求

见代码库：自定义拦截器com.heima.kafka.chapter2.ProducerInterceptorPrefix

/**
    * 自定义拦截器 
    */ 
public class ProducerInterceptorPrefix implements ProducerInterceptor<String, String> { 
    private volatile long sendSuccess = 0; 
    private volatile long sendFailure = 0; 
    
    @Override 
    public ProducerRecord<String, String> onSend( ProducerRecord<String, String> record) { 
        String modifiedValue = "prefix1-" + record.value(); 
        return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), record.key(), modifiedValue, record.headers()); 
// if (record.value().length() < 5) { 
// throw new RuntimeException();
// } 
// return record; 
    }
    
    @Override 
    public void onAcknowledgement( RecordMetadata recordMetadata, Exception e) { 
        if (e == null) { 
            sendSuccess++; 
        } else { 
            sendFailure++; 
        } 
    }
    
    @Override 
    public void close() { 
        double successRatio = (double) sendSuccess / (sendFailure + sendSuccess); 
        System.out.println("[INFO] 发送成功率=" + String.format("%f", successRatio * 100) + "%"); 
    }
    
    @Override 
    public void configure(Map<String, ?> map) { 
    } 
}

实现自定义拦截器之后需要在配置参数中指定这个拦截器，此参数的默认值为空，如下：

// 自定义拦截器使用 
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,ProducerDefineSerializer.cla ss.getName());

功能演示：

发送端

接收端

参考资料：《Kafka技术手册》
免费获取方式：私信【资料】免费获取
还有更多Java PDF学习资料等你来拿！！！

kafka-第二章-生产者
学习大纲一、kafka java客户端数据生产流程解析一、发送类型 1、同步发送通过send()发送完消息后...
Kafka Java客户端数据生产流程解析，从发送类型实现代码到
一、消息发送 1.Kafka Java客户端数据生产流程解析首先要构造一个 ProducerRecord 对象，...
Kafka高级特性解析(一)
Kafka高级特性解析(一) 生产者消息发送数据生产流程解析 Producer创建时，会创建一个Sender线...
哈哈
客户端发送数据实现： [java]viewplaincopy protectedvoidconnectServer...
Spark Structured Streaming 解析 JS
Producer 发送 JSON 数据到 Kafka： JSON 数据内容如下：代码 JDBCSink 写数据库...
Kafka 生产者调优
讲解样例代码中的参数含义： buffer.memory 参数是什么意思？ Kafka的客户端发送数据到服务器，一般...
Kafka学习——生产者和消费者客户端
kafka客户端 Kafka除了提供内置Java客户端外，还提供了二进制连接协议，即向Kafka网络端口发送适当的...
Python requests 爬取淘宝商品数据，并连接数据库，
前言开发环境 python 3.8pycharm 2021.2 专业版代码实现发送请求获取数据解析数据(筛选...
0005.ObjectMapper解析JSON(数组)字符串
要解析的JSON(数组)电文：解析代码问题1:默认不支持java.time.LocalDate数据类型的解析 ...
2网络协议基础
Web网络协议使用Java建立TCP连接并发送数据让飞秋死个机：利用 Java代码模拟客户端向服务器发送500...