（）1序列化(key.serializer 和 value.serializer 参数设置，和自定义序列化器)
kafka 自带String、Double、Integer、Long、Bytes、ByteBuffer、ByteArray等这些类型的序列化器，他们都实现了org.apache.kafka.common.serialization.Serializer 接口，这个接口有三个方法：
void configure(Map<String,?> configs, boolean isKey); // 用来配置当前类 ，主要在KafkaProducer 创建的时候调用，主要用来确认编码类型，默认为UTF-8 一般也不修改,并且key.serializer.encoding, value.serializer.encoding 和 serializer.encoding 这几个参数也不会配置
如StringSerializer:
byte[] serialize(String topic, T data);//序列化方法
void close();//用来关闭序列化 一般这个方法是空的，如果实现的话 主意幂等，因为它会被KafkaProducer调用多次
自定义序列化器：
2分区器(用于指定消息发送到主题的哪个分区)
producer 发送消息到broker 需要需要经过拦截器，序列化器，分区器 的一系列作用之后才能发往broker，如果发送消息的时候指定了分区，就不需要走分区器了。 分区是根据消息的key来指定的，相同的key 会发送到相同的分区
默认的分区器是org.apache.kafka.clients.producer.internals.DefaultPartitioner他们都实现了Partitioner接口 接口中有两个方法
int partition(String var1, Object var2, byte[] var3, Object var4, byte[] var5, Cluster var6);//返回int值，表示分区数
void close();//关闭分区时回收一些资源
Partitioner还有一个父接口Configurable 里面只有一个方法void configure(Map var1) 用来获取配置信息和初始化数据。
DefaultPartitioner（默认分区器）close 方法是空的，partition方法中定义了分区的逻辑，如果key不是空的会对key进行哈希等操作，如果key是空的则会轮训发送到topic中的各个分区中。
使用分区器 需要在producer 的配置中添加
3拦截器(properties.put(ProducerConfig .INTERCEPTOR_CLASSES_ CONFIG ,ProducerinterceptorPrefix.class.getName()); 可以设置多个拦截器，逗号隔开，会按照写的顺序执行拦截器)
需要实现ProducerInterceptor 接口 接口中有三个方法
ProducerRecord<K,V> onSend(ProducerRecord<K,V> var1);//拦截消息，对消息定制化操作，一般不会对key topic partition 进行修改，修改的话需要对其有准确的判断，不然会对分区计算和日志压缩有影响。
void onAcknowledgement(RecordMetadata var1, Exception var2);//会在被应答之前或者消息发送失败的时候调用 var2有值时 var1没有值，两个是互斥的，var2有值，则说明发送失败，没有值则说明发送成
void close();//用于关闭拦截器是回收一些资源 一般不用写
使用拦截器需要先配置producer
拦截器实现如下（统计发送成功率，在发送消息的时候拦截消息，在消息最前边添加prefixl- 字符串）
