在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

现在，您有了一个启动项目，该项目可以将异步对象发送到所需的主题。