目录
①、首先要构造一个 ProducerRecord 对象，该对象可以声明主题Topic、分区Partition、键 Key以及值 Value，主题和值是必须要声明的，分区和键可以不用指定。
②、调用send() 方法进行消息发送。
③、因为消息要到网络上进行传输，所以必须进行序列化，序列化器的作用就是把消息的 key 和 value对象序列化成字节数组。
④、接下来数据传到分区器，如果之间的 ProducerRecord 对象指定了分区，那么分区器将不再做任何事，直接把指定的分区返回；如果没有，那么分区器会根据 Key 来选择一个分区，选择好分区之后，生产者就知道该往哪个主题和分区发送记录了。
⑤、接着这条记录会被添加到一个记录批次里面，这个批次里所有的消息会被发送到相同的主题和分区。会有一个独立的线程来把这些记录批次发送到相应的 Broker 上。
③、Broker成功接收到消息，表示发送成功，返回消息的元数据（包括主题和分区信息以及记录在分区里的偏移量）。发送失败，可以选择重试或者直接抛出异常。
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.0.0</version></dependency>
实例代码：
1 package com.ys.utils; 2 3 import org.apache.kafka.clients.producer.*; 4 import java.util.Properties; 5 6 /** 7 * Create by YSOcean 8 */ 9 public class KafkaProducerUtils { 10 11 public static void main(String[] args) { 12 Properties kafkaProperties = new Properties(); 13 //配置broker地址信息14 kafkaProperties.put("bootstrap.servers", "192.168.146.200:9092,192.168.146.201:9092,192.168.146.202:9092"); 15 //配置 key 的序列化器16 kafkaProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 17 //配置 value 的序列化器18 kafkaProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 19 20 //通过上面的配置文件生成 Producer 对象21 Producer producer = new KafkaProducer(kafkaProperties); 22 //生成 ProducerRecord 对象，并制定 Topic，key 以及 value23 ProducerRecord<String,String> record =24 new ProducerRecord<String, String>("testTopic","key1","hello Producer"); 25 //发送消息26 producer.send(record); 27 } 28 }
通过运行上述代码，我们向名为 testTopic 的主题中发送了一条键为 key1，值为 hello Producer 的消息。
其中接口 serialization：
View Code②、自定义序列化器
如果Kafka提供的几个默认序列化器不能满足要求，即发送到 Kafka 的消息不是简单的字符串或整型，那么我们可以自定义序列化器。
比如对于如下的实体类 Person：
