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

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 以及 valueProducerRecord<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> {@Overridepublic void configure(Map configs, boolean isKey) {}

@Overridepublic 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];}

@Overridepublic 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<

【一线大厂Java面试题解析+核心总结学习笔记+最新架构讲解视频+实战项目源码讲义】
浏览器打开：qq.cn.hn/FTf 免费领取

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

@Overridepublic 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;}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}}

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