品牌网站建设优化公司,怎么建立一个网站卖东西,广告设计公司经营范围,国外免费建站网站搭建目录
1、消息生产流程
2、生产者常见参数配置
3、序列化器
基本概念
自定义序列化器
4、分区器
默认分区规则
自定义分区器
5、生产者拦截器
作用
自定义拦截器
6、生产者原理解析 1、消息生产流程 2、生产者常见参数配置 3、序列化器
基本概念
在Kafka中保存的数…目录
1、消息生产流程
2、生产者常见参数配置
3、序列化器
基本概念
自定义序列化器
4、分区器
默认分区规则
自定义分区器
5、生产者拦截器
作用
自定义拦截器
6、生产者原理解析 1、消息生产流程 2、生产者常见参数配置 3、序列化器
基本概念
在Kafka中保存的数据都是字节数组。消息发送前需要将消息序列化为字节数组进行发送。生产者通过key.serializer和value.serializer指定key和value的序列化器。Kafka使用org.apache.kafka.common.serialization.Serializer接口定义序列化器。Kafka已实现的序列化器有ByteArraySerializer、ByteBufferSerializer、BytesSerializer、DoubleSerializer、FloatSerializer、IntegerSerializer、StringSerializer、LongSerializer、ShortSerializer。
自定义序列化器
实现org.apache.kafka.common.serialization.SerializerT接口并实现其中的serializer方法。
Data
public class User {private Integer userId;private String username;
}public class UserSerializer implements SerializerUser {Overridepublic void configure(MapString, ? configs, boolean isKey) {// do nothing}Overridepublic byte[] serialize(String topic, User data) {try {// 如果数据是null则返回nullif (data null) return null;Integer userId data.getUserId();String username data.getUsername();int length 0;byte[] bytes null;if (null ! username) {bytes username.getBytes(utf-8);length bytes.length;}// 第一个4字节存储userId的值// 第二个4字节存储username字节数组的长度int值// 第三个length长度存储username序列化之后的字节数组ByteBuffer buffer ByteBuffer.allocate(4 4 length);buffer.putInt(userId);buffer.putInt(length);buffer.put(bytes);return buffer.array();} catch (UnsupportedEncodingException e) {throw new SerializationException(序列化数据异常);}}Overridepublic void close() {// do nothing}
}
4、分区器
默认分区规则
KafkaProducer.partition()DefaultPartitioner.partition()
如果record提供了分区号则使⽤record提供的分区号如果record没有提供分区号则使⽤key的序列化后的值的hash值对分区数量取模如果record没有提供分区号也没有提供key则使⽤轮询的⽅式分配分区号。
自定义分区器
实现org.apache.kafka.clients.producer.Partitioner接口并实现其中的partition方法。
在生产者参数中通过配置partitioner.class指定自定义分区器。
/*** 自定义分区器*/
public class MyPartitioner implements Partitioner {Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 此处可以计算分区的数字。// 我们直接指定为2return 2;}Overridepublic void close() {}Overridepublic void configure(MapString, ? configs) {}
}5、生产者拦截器
作用 在发送消息前或者在执行回调逻辑前对消息做一些定制化的处理比如修改消息打印消息日志等。此外Producer允许设置多个拦截器从而形成一条拦截器链Producer将按照指定顺序调用它们。
自定义拦截器 自定义拦截器实现org.apache.kafka.clients.producer.ProducerInterceptor接口并实现其中的onSend()、onAcknowledgement()、close()接口。其中
onSend(ProducerRecord)Producer 确保在消息被序列化前调⽤该⽅法。⽤户可以在该⽅法中对消息做任何操作但最好保证不要修 改消息所属的topic和分区否则会影响⽬标分区的计算。onAcknowledgement(RecordMetadata, Exception)该⽅法会在消息被应答之前或消息发送失败时调⽤ 并且通常都是在Producer回调逻辑触发之前。close关闭Interceptor主要⽤于执⾏⼀些资源清理⼯作。 在生产者参数中通过配置ProducerConfig.INTERCEPTOR_CLASSES_CONFIG指定自定义拦截器。
public class InterceptorKEY, VALUE implements ProducerInterceptorKEY, VALUE {private static final Logger LOGGER LoggerFactory.getLogger(InterceptorTwo.class);Overridepublic ProducerRecordKEY, VALUE onSend(ProducerRecordKEY, VALUE record) {System.out.println(拦截器---go);// 此处根据业务需要对相关的数据作修改String topic record.topic();Integer partition record.partition();Long timestamp record.timestamp();KEY key record.key();VALUE value record.value();Headers headers record.headers();// 添加消息头headers.add(interceptor, interceptor.getBytes());ProducerRecordKEY, VALUE newRecord new ProducerRecordKEY, VALUE(topic, partition, timestamp, key, value, headers);return newRecord;}Overridepublic void onAcknowledgement(RecordMetadata metadata, Exception exception) {System.out.println(拦截器---back);if (exception ! null) {// 如果发生异常记录在日志中LOGGER.error(exception.getMessage());}}Overridepublic void close() {}Overridepublic void configure(MapString, ? configs) {}
}
6、生产者原理解析 以上内容为个人学习理解如有问题欢迎在评论区指出。 部分内容截取自网络如有侵权联系作者删除。
