kafka异步发送数据_在Kafka上异步发送数据

古城微笑少年丶 2023-02-25 04:59 99阅读 0赞

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 KafkaTemplate<String, 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) { 
             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（）对我而言并不真正有意义。 但是onFailure（）我可以记录该异常，以便通知我我的kafka服务器是否有问题。

我还要与您分享另一件事。 为了通过kafkatemplate发送对象，我必须为其配备序列化文件。

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定义序列化器。 最简单的方法之一是将其添加到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.html][https_www.javacodegeeks.com_2020_01_send-your-data-async-on-kafka.html]

kafka异步发送数据

[https_www.javacodegeeks.com_2020_01_send-your-data-async-on-kafka.html]: https://www.javacodegeeks.com/2020/01/send-your-data-async-on-kafka.html