kafka異步發(fā)送數(shù)據(jù)

對(duì)于一個(gè)項(xiàng)目，我試圖記錄用戶的基本交易，例如添加和刪除一個(gè)項(xiàng)目以及多種類型的項(xiàng)目，并為每筆交易向kafka發(fā)送一條消息。 日志機(jī)制的準(zhǔn)確性不是至關(guān)重要的，在kafka服務(wù)器停機(jī)的情況下，我不希望它阻止我的業(yè)務(wù)代碼。 在這種情況下，將數(shù)據(jù)發(fā)送到kafka的異步方法是更好的方法。

我的kafka生產(chǎn)者代碼在其引導(dǎo)項(xiàng)目中。 為了使其異步，我只需要添加兩個(gè)注釋：@EnableAsync和@Async。

@EnableAsync將在您的配置類中使用（還要記住，帶有@SpringBootApplication的類也是配置類），并將嘗試查找TaskExecutor bean。 如果沒(méi)有，它將創(chuàng)建一個(gè)SimpleAsyncTaskExecutor。 SimpleAsyncTaskExecutor適用于玩具項(xiàng)目，但對(duì)于任何大于此的項(xiàng)目都存在一定的風(fēng)險(xiǎn)，因?yàn)樗幌拗撇l(fā)線程，也不會(huì)重用線程。 為了安全起見(jiàn)，我們還將添加一個(gè)任務(wù)執(zhí)行者bean。

所以，

@SpringBootApplication public class KafkaUtilsApplication { public static void main(String[] args) { SpringApplication.run(KafkaUtilsApplication. class , args); } }

會(huì)變成

@EnableAsync @SpringBootApplication public class KafkaUtilsApplication { public static void main(String[] args) { SpringApplication.run(KafkaUtilsApplication. class , args); } @Bean public Executor taskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize( 2 ); executor.setMaxPoolSize( 2 ); executor.setQueueCapacity( 500 ); executor.setThreadNamePrefix( "KafkaMsgExecutor-" ); executor.initialize(); return executor; } }

如您所見(jiàn)，這里沒(méi)有太多變化。 我設(shè)置的默認(rèn)值應(yīng)根據(jù)您的應(yīng)用程序需求進(jìn)行調(diào)整。

我們需要的第二件事是添加@Async。

我的舊代碼是：

@Service public class KafkaProducerServiceImpl implements KafkaProducerService { private static final String TOPIC = "logs" ; @Autowired private KafkaTemplate<String, KafkaInfo> kafkaTemplate; @Override public void sendMessage(String id, KafkaType kafkaType, KafkaStatus kafkaStatus) { kafkaTemplate.send(TOPIC, new KafkaInfo(id, kafkaType, kafkaStatus); } }

如您所見(jiàn)，同步代碼非常簡(jiǎn)單。 它只需要kafkaTemplate并將消息對(duì)象發(fā)送到“ logs”主題。 我的新代碼比這更長(zhǎng)。

@Service public class KafkaProducerServiceImpl implements KafkaProducerService { private static final String TOPIC = "logs" ; @Autowired private KafkaTemplate kafkaTemplate; @Async @Override public void sendMessage(String id, KafkaType kafkaType, KafkaStatus kafkaStatus) { ListenableFuture<SendResult<String, KafkaInfo>> future = kafkaTemplate.send(TOPIC, new KafkaInfo(id, kafkaType, kafkaStatus)); future.addCallback( new ListenableFutureCallback<>() { @Override public void onSuccess( final SendResult<String, KafkaInfo> message) { // left empty intentionally } @Override public void onFailure( final Throwable throwable) { // left empty intentionally } }); } }

在這里，onSuccess（）對(duì)我而言并不真正有意義。 但是onFailure（）我可以記錄該異常，以便通知我我的kafka服務(wù)器是否有問(wèn)題。

我還要與您分享另一件事。 為了通過(guò)kafkatemplate發(fā)送對(duì)象，我必須為其配備序列化文件。

public class KafkaInfoSerializer implements Serializer<kafkainfo> { @Override public void configure(Map map, boolean b) { } @Override public byte [] serialize(String arg0, KafkaInfo info) { byte [] retVal = null ; ObjectMapper objectMapper = new ObjectMapper(); try { retVal = objectMapper.writeValueAsString(info).getBytes(); } catch (Exception e) { // log the exception } return retVal; } @Override public void close() { } }

另外，不要忘記為其添加配置。 有幾種方法可以為kafka定義序列化器。 最簡(jiǎn)單的方法之一是將其添加到application.properties。

spring.kafka.producer.key-serializer = org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer = com.sezinkarli.kafkautils.serializer.KafkaInfoSerializer

現(xiàn)在，您有了一個(gè)啟動(dòng)項(xiàng)目，該項(xiàng)目可以將異步對(duì)象發(fā)送到所需的主題。

翻譯自: https://www.javacodegeeks.com/2020/01/send-your-data-async-on-kafka.html

kafka異步發(fā)送數(shù)據(jù)

總結(jié)

以上是生活随笔為你收集整理的kafka异步发送数据_在Kafka上异步发送数据的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。