10分钟学会如何在SpringBoot程序中使用Kafka作为消息队列

x33g5p2x  于2020-09-08 发布在 Kafka  
字(8.2k)|赞(0)|评价(0)|浏览(764)

Step1:创建项目

直接通过Spring 官方提供的 Spring Initializr 创建或者直接使用 IDEA 创建皆可。

Step2: 配置 Kafka

通过 application.yml 配置文件配置 Kafka 基本信息

  1. server:
  2.   port: 9090
  4. spring:
  5.   kafka:
  6.     consumer:
  7.       bootstrap-servers: localhost:9092
  8.       # 配置消费者消息offset是否自动重置(消费者重连会能够接收最开始的消息)
  9.       auto-offset-reset: earliest
  10.     producer:
  11.       bootstrap-servers: localhost:9092
  12.       # 发送的对象信息变为json格式
  13.       value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
  14. kafka:
  15.   topic:
  16.     my-topic: my-topic
  17.     my-topic2: my-topic2

Kafka 额外配置类：

  1. package cn.javaguide.springbootkafka01sendobjects.config;
  3. import org.apache.kafka.clients.admin.NewTopic;
  4. import org.springframework.beans.factory.annotation.Value;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  7. import org.springframework.kafka.support.converter.RecordMessageConverter;
  8. import org.springframework.kafka.support.converter.StringJsonMessageConverter;
  10. /**
  11.  * @author shuang.kou
  12.  */
  13. @Configuration
  14. public class KafkaConfig {
  16.     @Value("${kafka.topic.my-topic}")
  17.     String myTopic;
  18.     @Value("${kafka.topic.my-topic2}")
  19.     String myTopic2;
  21.     /**
  22.      * JSON消息转换器
  23.      */
  24.     @Bean
  25.     public RecordMessageConverter jsonConverter() {
  26.         return new StringJsonMessageConverter();
  27.     }
  29.     /**
  30.      * 通过注入一个 NewTopic 类型的 Bean 来创建 topic，如果 topic 已存在，则会忽略。
  31.      */
  32.     @Bean
  33.     public NewTopic myTopic() {
  34.         return new NewTopic(myTopic, 2, (short) 1);
  35.     }
  37.     @Bean
  38.     public NewTopic myTopic2() {
  39.         return new NewTopic(myTopic2, 1, (short) 1);
  40.     }
  41. }

当我们到了这一步之后，你就可以试着运行项目了，运行成功后你会发现 Spring Boot 会为你创建两个topic:

  1. my-topic: partition 数为 2, replica 数为 1
  2. my-topic2:partition 数为 1, replica 数为 1

通过上一节说的：kafka-topics --describe --zookeeper zoo1:2181 命令查看或者直接通过IDEA 提供的 Kafka 可视化管理插件-Kafkalytic 来查看

Step3:创建要发送的消息实体类

  1. package cn.javaguide.springbootkafka01sendobjects.entity;
  3. public class Book {
  4.     private Long id;
  5.     private String name;
  7.     public Book() {
  8.     }
  10.     public Book(Long id, String name) {
  11.         this.id = id;
  12.         this.name = name;
  13.     }
  15.     省略 getter/setter以及 toString方法
  16. }

Step4:创建发送消息的生产者

这一步内容比较长，会一步一步优化生产者的代码。

  1. import org.slf4j.Logger;
  2. import org.slf4j.LoggerFactory;
  3. import org.springframework.kafka.core.KafkaTemplate;
  4. import org.springframework.stereotype.Service;
  6. @Service
  7. public class BookProducerService {
  9.     private static final Logger logger = LoggerFactory.getLogger(BookProducerService.class);
  11.     private final KafkaTemplate<String, Object> kafkaTemplate;
  13.     public BookProducerService(KafkaTemplate<String, Object> kafkaTemplate) {
  14.         this.kafkaTemplate = kafkaTemplate;
  15.     }
  17.     public void sendMessage(String topic, Object o) {
  18.         kafkaTemplate.send(topic, o);
  19.     }
  20. }

我们使用Kafka 提供的 KafkaTemplate 调用 send()方法出入要发往的topic和消息内容即可很方便的完成消息的发送:

  1.   kafkaTemplate.send(topic, o);

如果我们想要知道消息发送的结果的话，sendMessage方法这样写：

  1.     public void sendMessage(String topic, Object o) {
  2.         try {
  3.             SendResult<String, Object> sendResult = kafkaTemplate.send(topic, o).get();
  4.             if (sendResult.getRecordMetadata() != null) {
  5.                 logger.info("生产者成功发送消息到" + sendResult.getProducerRecord().topic() + "-> " + sendResult.getProducerRecord().value().toString());
  6.             }
  7.         } catch (InterruptedException | ExecutionException e) {
  8.             e.printStackTrace();
  9.         }
  10.     }

但是这种属于同步的发送方式并不推荐，没有利用到 Future对象的特性。

KafkaTemplate 调用 send()方法实际上返回的是ListenableFuture 对象。

send()方法源码如下：

  1. 	@Override
  2. 	public ListenableFuture<SendResult<K, V>> send(String topic, @Nullable V data) {
  3. 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, data);
  4. 		return doSend(producerRecord);
  5. 	}

ListenableFuture 是Spring提供了继承自Future 的接口。

ListenableFuture方法源码如下：

  1. public interface ListenableFuture<T> extends Future<T> {
  2.     void addCallback(ListenableFutureCallback<? super T> var1);
  4.     void addCallback(SuccessCallback<? super T> var1, FailureCallback var2);
  6.     default CompletableFuture<T> completable() {
  7.         CompletableFuture<T> completable = new DelegatingCompletableFuture(this);
  8.         this.addCallback(completable::complete, completable::completeExceptionally);
  9.         return completable;
  10.     }
  11. }

继续优化sendMessage方法

  1.     public void sendMessage(String topic, Object o) {
  3.         ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, o);
  4.         future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
  6.             @Override
  7.             public void onSuccess(SendResult<String, Object> sendResult) {
  8.                 logger.info("生产者成功发送消息到" + topic + "-> " + sendResult.getProducerRecord().value().toString());
  9.             }
  10.             @Override
  11.             public void onFailure(Throwable throwable) {
  12.                 logger.error("生产者发送消息：{} 失败，原因：{}", o.toString(), throwable.getMessage());
  13.             }
  14.         });
  15.     }

使用lambda表达式再继续优化：

  1.     public void sendMessage(String topic, Object o) {
  3.         ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, o);
  4.         future.addCallback(result -> logger.info("生产者成功发送消息到topic:{} partition:{}的消息", result.getRecordMetadata().topic(), result.getRecordMetadata().partition()),
  5.                 ex -> logger.error("生产者发送消失败，原因：{}", ex.getMessage()));
  6.     }

再来简单研究一下 send(String topic, @Nullable V data) 方法。

我们使用send(String topic, @Nullable V data)方法的时候实际会new 一个ProducerRecord对象发送，

  1. 	@Override
  2. 	public ListenableFuture<SendResult<K, V>> send(String topic, @Nullable V data) {
  3. 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, data);
  4. 		return doSend(producerRecord);
  5. 	}

ProducerRecord类中有多个构造方法:

  1.    public ProducerRecord(String topic, V value) {
  2.         this(topic, null, null, null, value, null);
  3.     }
  4.     public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V 
  5.         ......
  6.     }

如果我们想在发送的时候带上timestamp（时间戳）、key等信息的话，sendMessage()方法可以这样写：

  1.     public void sendMessage(String topic, Object o) {
  2.       // 分区编号最好为 null，交给 kafka 自己去分配
  3.         ProducerRecord<String, Object> producerRecord = new ProducerRecord<>(topic, null, System.currentTimeMillis(), String.valueOf(o.hashCode()), o);
  4.       
  5.         ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(producerRecord);
  6.         future.addCallback(result -> logger.info("生产者成功发送消息到topic:{} partition:{}的消息", result.getRecordMetadata().topic(), result.getRecordMetadata().partition()),
  7.                 ex -> logger.error("生产者发送消失败，原因：{}", ex.getMessage()));
  8.     }

Step5:创建消费消息的消费者

通过在方法上使用 @KafkaListener 注解监听消息，当有消息的时候就会通过 poll 下来消费。

  1. import cn.javaguide.springbootkafka01sendobjects.entity.Book;
  2. import com.fasterxml.jackson.core.JsonProcessingException;
  3. import com.fasterxml.jackson.databind.ObjectMapper;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.beans.factory.annotation.Value;
  8. import org.springframework.kafka.annotation.KafkaListener;
  9. import org.springframework.stereotype.Service;
  11. @Service
  12. public class BookConsumerService {
  14.     @Value("${kafka.topic.my-topic}")
  15.     private String myTopic;
  16.     @Value("${kafka.topic.my-topic2}")
  17.     private String myTopic2;
  18.     private final Logger logger = LoggerFactory.getLogger(BookProducerService.class);
  19.     private final ObjectMapper objectMapper = new ObjectMapper();
  22.     @KafkaListener(topics = {"${kafka.topic.my-topic}"}, groupId = "group1")
  23.     public void consumeMessage(ConsumerRecord<String, String> bookConsumerRecord) {
  24.         try {
  25.             Book book = objectMapper.readValue(bookConsumerRecord.value(), Book.class);
  26.             logger.info("消费者消费topic:{} partition:{}的消息 -> {}", bookConsumerRecord.topic(), bookConsumerRecord.partition(), book.toString());
  27.         } catch (JsonProcessingException e) {
  28.             e.printStackTrace();
  29.         }
  30.     }
  32.     @KafkaListener(topics = {"${kafka.topic.my-topic2}"}, groupId = "group2")
  33.     public void consumeMessage2(Book book) {
  34.         logger.info("消费者消费{}的消息 -> {}", myTopic2, book.toString());
  35.     }
  36. }

Step6:创建一个 Rest Controller

  1. import cn.javaguide.springbootkafka01sendobjects.entity.Book;
  2. import cn.javaguide.springbootkafka01sendobjects.service.BookProducerService;
  3. import org.springframework.beans.factory.annotation.Value;
  4. import org.springframework.web.bind.annotation.PostMapping;
  5. import org.springframework.web.bind.annotation.RequestMapping;
  6. import org.springframework.web.bind.annotation.RequestParam;
  7. import org.springframework.web.bind.annotation.RestController;
  9. import java.util.concurrent.atomic.AtomicLong;
  11. /**
  12.  * @author shuang.kou
  13.  */
  14. @RestController
  15. @RequestMapping(value = "/book")
  16. public class BookController {
  17.     @Value("${kafka.topic.my-topic}")
  18.     String myTopic;
  19.     @Value("${kafka.topic.my-topic2}")
  20.     String myTopic2;
  21.     private final BookProducerService producer;
  22.     private AtomicLong atomicLong = new AtomicLong();
  24.     BookController(BookProducerService producer) {
  25.         this.producer = producer;
  26.     }
  28.     @PostMapping
  29.     public void sendMessageToKafkaTopic(@RequestParam("name") String name) {
  30.         this.producer.sendMessage(myTopic, new Book(atomicLong.addAndGet(1), name));
  31.         this.producer.sendMessage(myTopic2, new Book(atomicLong.addAndGet(1), name));
  32.     }
  33. }

Step7:测试

输入命令：

  1. curl -X POST -F 'name=Java' http://localhost:9090/book

内容来源于网络，如有侵权，请联系作者删除！

kafkaspringboot消息队列

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com