专门做旅游的视频网站,建设网站盈利2015,青岛做教育的网站,建设监理协会官方网站kafka异步发送数据对于一个项目#xff0c;我试图记录用户的基本交易#xff0c;例如添加和删除一个项目以及多种类型的项目#xff0c;并为每笔交易向kafka发送一条消息。 日志机制的准确性不是至关重要的#xff0c;在kafka服务器停机的情况下#xff0c;我不希望它阻止… kafka异步发送数据 对于一个项目我试图记录用户的基本交易例如添加和删除一个项目以及多种类型的项目并为每笔交易向kafka发送一条消息。 日志机制的准确性不是至关重要的在kafka服务器停机的情况下我不希望它阻止我的业务代码。 在这种情况下将数据发送到kafka的异步方法是更好的方法。 我的kafka生产者代码在其引导项目中。 为了使其异步我只需要添加两个注释EnableAsync和Async。 EnableAsync将在您的配置类中使用还要记住带有SpringBootApplication的类也是配置类并将尝试查找TaskExecutor bean。 如果没有它将创建一个SimpleAsyncTaskExecutor。 SimpleAsyncTaskExecutor适用于玩具项目但对于任何大于此的项目都存在一定的风险因为它不限制并发线程也不会重用线程。 为了安全起见我们还将添加一个任务执行者bean。 所以 SpringBootApplication public class KafkaUtilsApplication { public static void main(String[] args) { SpringApplication.run(KafkaUtilsApplication. class , args); } } 会变成 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; } } 如您所见这里没有太多变化。 我设置的默认值应根据您的应用程序需求进行调整。 我们需要的第二件事是添加Async。 我的旧代码是 Service public class KafkaProducerServiceImpl implements KafkaProducerService { private static final String TOPIC logs ; Autowired private KafkaTemplateString, KafkaInfo kafkaTemplate; Override public void sendMessage(String id, KafkaType kafkaType, KafkaStatus kafkaStatus) { kafkaTemplate.send(TOPIC, new KafkaInfo(id, kafkaType, kafkaStatus); } } 如您所见同步代码非常简单。 它只需要kafkaTemplate并将消息对象发送到“ logs”主题。 我的新代码比这更长。 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) { ListenableFutureSendResultString, KafkaInfo future kafkaTemplate.send(TOPIC, new KafkaInfo(id, kafkaType, kafkaStatus)); future.addCallback( new ListenableFutureCallback() { Override public void onSuccess( final SendResultString, KafkaInfo message) { // left empty intentionally } Override public void onFailure( final Throwable throwable) { // left empty intentionally } }); } } 在这里onSuccess对我而言并不真正有意义。 但是onFailure我可以记录该异常以便通知我我的kafka服务器是否有问题。 我还要与您分享另一件事。 为了通过kafkatemplate发送对象我必须为其配备序列化文件。 public class KafkaInfoSerializer implements Serializerkafkainfo { 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定义序列化器。 最简单的方法之一是将其添加到application.properties。 spring.kafka.producer.key-serializer org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer com.sezinkarli.kafkautils.serializer.KafkaInfoSerializer 现在您有了一个启动项目该项目可以将异步对象发送到所需的主题。 翻译自: https://www.javacodegeeks.com/2020/01/send-your-data-async-on-kafka.htmlkafka异步发送数据
