您的位置:首页 > 娱乐 > 明星 > Kafka(二)Producer第一篇

Kafka(二)Producer第一篇

2024/12/23 16:09:26 来源:https://blog.csdn.net/jiezheee/article/details/140262319  浏览:    关键词:Kafka(二)Producer第一篇

一,Client开发


生产逻辑需要具备以下几个 步骤:
(1)配置生产者客户端参数及创建相应的生产者实例。
(2)构建待发送的消息。
(3)发送消息。
(4)关闭生产者实例。
public class Producer {private static final String BROKER_LIST = "localhost:9092";private static final String TOPIC = "TOPIC-A";public static Properties initConfig() {Properties properties = new Properties();// 以下3个必须properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 客户端IDproperties.put(ProducerConfig.CLIENT_ID_CONFIG, "eris-kafka-producer");// 重试次数properties.put(ProducerConfig.RETRIES_CONFIG, 3);// 成功收到消息分区副本数,默认为1,即leader副本收到就返回成功。注意是字符串!properties.put(ProducerConfig.ACKS_CONFIG, "1");// 生产者客户端能发送的消息的最大值,单位为B,默认1Mproperties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,  1048576);// Producer等待请求响应的最长时间,单位ms,默认值为30000properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,  30000);// 生产者发送ProducerBatch之前等待更多ProducerRecord加入的时间。默认为0,ProducerBatch被填满时发出properties.put(ProducerConfig.LINGER_MS_CONFIG,  0);return properties;}public static ProducerRecord<String, String> initMessage() {return new ProducerRecord<>(TOPIC, "hi eris");}public static void main(String[] args) {Properties properties = initConfig();KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);ProducerRecord record = initMessage();kafkaProducer.send(record);kafkaProducer.close();}
}
注意:KafkaProducer,它是线程安全的,我们可以在多线程的环境中复用它。
消费者客户端KafkaConsumer是非线程安全的。
1,发送消息的构造
消息对象是ProducerRecord,业务信息是value字段。
  • key可以让消息再二次归类,同一个key的消息会被划分到同一个分区中;
  • topic属性和value属性是必填项,其余属性是选填项;
2,主要Producer参数
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG并非需要所有的broker地址,生产者会从给定的broker里查找到其他broker的信息,一般至少要2个。
3, 消息的发送
3-1,消息发送的3种模式:发后即忘、同步及异步
A,发后即忘:
kafkaProducer.send(record);
B,同步:
kafkaProducer.send(record).get();
C,异步:
  • 不推荐手动用future实现业务异步,诸多消息对应的Future对象的处理会引起逻辑的混乱。
  • 对于同一个分区,如果消息record1于record2之前先发送,那么KafkaProducer可以保证对应的callback1在callback2之前调用,也就是说 ,回调函数的调用也可以保证分区有序。
3-2,KafkaProducer中一般会发生两种类型的异常:可重试的异常和不可重试的异常。
对于可重试异常,如果配置了 retries 参数,那么只要在规定的重试次数内自行恢复了,就不会抛出异常;不可重试异常会直接抛出异常。
3-3,close方法会回收资源,并且阻塞等待之前所有的发送请求完成后再关闭 KafkaProducer,可带timeout。
3-4,消息在通过send发往broker的过程中,有可能需要经过拦截器(Interceptor)、序列化器(Serializer)和分区器(Partitioner)才能被真正地发往 broker。
4,序列化
生产者需要用序列化器(Serializer) 把对象转换成字节数组才能通过网络发送给Kafka。
如果 Kafka 客户端提供的几种序列化器都无法满足需求,则可以选择使用如Avro、JSON、Thrift、ProtoBuf和Protostuff等通用的序列化工具来实现,或者使用 自定义的序列化器来实现
5,分区器
如果ProducerRecord中指定了partition字段,那么就不需要分区器的作用。
反之若没有指定partition字段,那么就需要依赖分区器, 根据key这个字段来计算partition的值。
  • 如果key不为null,计算得到的分区会是所有分区中的任意一个;
  • 如果key为null,计算得到的分区号仅为可用分区中的任意一个;
一旦主题中增加了分区,那么就难以保证key与分区之间的映射关系了。
可以自定义分区器。
6,拦截器
自定义拦截器需要实现org.apache.kafka.clients.producer.ProducerInterceptor接口,接口中包含3个方法:
  • onSend:将消息序列化和计算分区之前调用;
  • onAcknowledgement:消息被应答(ack)之前或消息发送失败时调用,优先于用户设定的 Callback之前执行;
  • interceptor.classes配置拦截器链(逗号分隔),拦截链会按照参数配置的顺序一一执行。如果某个拦截器执行失败,下一个拦截器会接 着上一个执行成功的拦截器继续执行。

二,重要参数


1.acks
用来指定分区中必须要有多少个副本收到这条消息,之后生产者才会认为这条消息是成功写入的。
  • ack=1(默认),leader写入(落盘)即成功;
  • ack=0,发送后就认为成功,不需要等待任何服务端的响应;
  • ack=-1或者ack=“all”,需要ISR中的所有副本都成功写入;
2,retries
生产者重试的次数,默认值为0,即不重试。
保证消息顺序
max.in.flight.requests.per.connection
该参数指定了生产者在收到服务器响应之前可以发送多少个消息(即最多缓存的请求数)。它的值越高,就会占用越多的内存,不过也会提升吞吐量。把它设为 1 可以保证消息是按照发送的顺序写入服务器的,即使发生了重试。
如果把 retries 设为非零整数,同时把 max.in.flight.requests.per.connection 设为比 1 大的数,那么,如果第一个批次消息写入失败,而第二个批次写入成功,broker 会重试写入第一个批次。如果此时第一个批次也写入成功,那么两个批次的顺序就反过来了
一般来说,如果某些场景要求消息是有序的,那么消息是否写入成功也是很关键的,所以不建议把 retries 设为 0。可以把 max.in.flight.requests.per.connection 设为 1,这样在生产者尝试发送第一批消息时,就不会有其他的消息发送给 broker。不过这样会严重影响生产者的吞吐量,所以只有在对消息的顺序有严格要求的情况下才能这么做。
3,linger.ms
这个参数用来指定生产者发送ProducerBatch之前等待更多ProducerRecord加入ProducerBatch的时间,默认值为0,即生产者会在 ProducerBatch 被填满时发送出去。
其他:

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com