您的位置:首页 > 汽车 > 时评 > 湖南网站制作公司推荐_中国建筑股份有限公司 官网_世界杯比分_福州seo结算

湖南网站制作公司推荐_中国建筑股份有限公司 官网_世界杯比分_福州seo结算

2024/11/18 8:40:46 来源:https://blog.csdn.net/happycao123/article/details/142183454  浏览:    关键词:湖南网站制作公司推荐_中国建筑股份有限公司 官网_世界杯比分_福州seo结算
湖南网站制作公司推荐_中国建筑股份有限公司 官网_世界杯比分_福州seo结算

同步发送 or 异步发送

       消息发送根据是否需要处理发送的结果分为同步发送、异步发送。

同步发送:等待发送结果返回,这种方式是可靠的,因为异常能及时处理,但同步发送需要阻塞等待一条消息发送完才处理下一条,吞吐量差。


异步发送:发送是异步的,不关心发送的结果,吞吐量最高,但可能存在发送失败的情况。

    本质上kafka 客户端提供的发送接口都是异步的,因为发送接口返回的是一个Future对象。对于同步发送通过future.get获取发送结果。异步发送则忽略send 返回值。

ListenableFuture<SendResult> future = kafkaTemplate.send(topic, content);try {SendResult sendResult = future.get();} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}

发送完成回调

有没有办法既要异步发送还要能处理发送失败的场景,这就是第三种,发送完成时,执行相应的回调方法。这是折中方案,兼顾效率且保证发送失败能被监控到。

producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {if(e != null){System.out.println("send error ");
}else {System.out.println("send result  topic ="+recordMetadata.topic() + " partition=" + recordMetadata.partition() + "  offset=" + recordMetadata.offset() );
}}
});

发送异常

       有些发送异常可以通过重试几次后解决,比如网络异常,对于有些异常比如消息太大超出kafka配置的最大消息字节数,这类异常重试也会失败,所以这类异常KafkaProducer 不会进行任何重试。对于可重试异常可以配置重试次数

spring.kafka.producer.retries=10

SpringBoot 集成简单介绍

     参考上篇文章SpringBoot 集成配置(pom依赖、application配置),简单讲解SpringBoot 几个重要自动装配类。

KafkaAutoConfiguration

KafkaAutoConfiguration给我们自动配置了几个类

KafkaTemplate:可以通过KafkaTemplate进行发送消息,本质上内部还是使用的KafkaProducer发送消息的。

ProducerFactory:KafkaProducer工厂,通过createProducer()方法可以获取(KafkaProducer) 进行发送消息,避免直接new KafkaProducer

使用方式也很简单,由于直接KafkaAutoConfiguration已经定义了相关Bean, 使用时注入Bean即可

图片

@Autowired
private KafkaTemplate kafkaTemplate;@Autowired
private ProducerFactory producerFactory;

具体代码

同步发送、异步发送的方式直接使用 kafkaTemplate即可完成,同步发送结果处理:这里简单的打印出消息的topic partition offset 等信息如下图

ListenableFuture<SendResult> future = kafkaTemplate.send(topic, content);
SendResult sendResult = future.get();
RecordMetadata recordMetadata = sendResult.getRecordMetadata();
System.out.println("send result  topic ="+recordMetadata.topic() + " partition=" + recordMetadata.partition() + "  offset=" + recordMetadata.offset() );
 

                                   

图片

发送回调kafkaTemplate没有对应api , 需要通过Producer发送,我们通过producerFactory获取。

ProducerRecord record = new ProducerRecord(topic,content);Producer producer = producerFactory.createProducer();producer.send(record, new Callback() {@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if(e != null){System.out.println("send error ");}else {System.out.println("send result  topic ="+recordMetadata.topic() + " partition=" + recordMetadata.partition() + "  offset=" + recordMetadata.offset() );}}});

版权声明:

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

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