脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → spring boot 使用apache kafka

在Spring Boot应用程序中使用Apache Kafka的方法步骤详解

 更新时间：2018年11月05日 10:32:14   作者：banq  
这篇文章主要介绍了在Spring Boot应用程序中使用Apache Kafka的方法步骤详解,非常不错，具有一定的参考借鉴价值，需要的朋友可以参考下

第1步：生成我们的项目： Spring Initializr 来生成我们的项目。我们的项目将提供Spring MVC / Web支持和Apache Kafka支持。

第2步：发布/读取Kafka主题中的消息：

<b>public</b> <b>class</b> User {
  <b>private</b> String name;
  <b>private</b> <b>int</b> age;
  <b>public</b> User(String name, <b>int</b> age) {
    <b>this</b>.name = name;
    <b>this</b>.age = age;
  }
}

第3步：通过application.yml配置文件配置Kafka:

我们需要创建配置文件。我们需要以某种方式配置我们的Kafka生产者和消费者，以便能够发布和读取与主题相关的消息。相比建立一个使用@Configuration标注的Java类，我们可以直接使用配置文件application.properties或application.yml。Spring Boot让我们避免像过去一样编写的所有样板代码，同时为我们提供了更加智能的配置应用程序的方法，如下所示：

server: port: 9000
spring:
  kafka:
   consumer:
    bootstrap: localhost:9092
    group-id: group_id
    auto-offset-reset: earliest
    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
producer:
    bootstrap: localhost:9092
    key-serializer: org.apache.kafka.common.serialization.StringSerializer
    value-serializer: org.apache.kafka.common.serialization.StringSerializer

第4步：创建一个生产者，创建生产者会将我们的消息写入该主题。

<b>public</b> <b>class</b> Producer {
  <b>private</b> <b>static</b> <b>final</b> Logger logger = LoggerFactory.getLogger(Producer.<b>class</b>);
  <b>private</b> <b>static</b> <b>final</b> String TOPIC = <font>"users"</font><font>;
  @Autowired
  <b>private</b> KafkaTemplate<String, String> kafkaTemplate;
  <b>public</b> <b>void</b> sendMessage(String message) {
    logger.info(String.format(</font><font>"#### -> Producing message -> %s"</font><font>, message));
    <b>this</b>.kafkaTemplate.send(TOPIC, message);
  }
}
</font>

自动连接autowire到 KafkaTemplate ，使用它将消息发布到主题 - 这就是消息的生产者！

第5步：创建一个消费者，消费者是负责根据您自己的业务逻辑的需求阅读处理消息的消息的服务。要进行设置，请输入以下内容：

@Service
<b>public</b> <b>class</b> Consumer {

  <b>private</b> <b>final</b> Logger logger = LoggerFactory.getLogger(Producer.<b>class</b>);

  @KafkaListener(topics = <font>"users"</font><font>, groupId = </font><font>"group_id"</font><font>)
  <b>public</b> <b>void</b> consume(String message) throws IOException {
    logger.info(String.format(</font><font>"#### -> Consumed message -> %s"</font><font>, message));
  }
}
</font>

在这里，我们告诉我们的方法void consume（String message）订阅用户的主题，并将每条消息发送到应用程序日志。在您的实际应用程序中，您可以按照业务需要的方式处理消息。

第6步：创建REST控制器，们已经拥有了能够消费Kafka消息所需的全部内容。

为了充分展示我们创建的所有内容的工作原理，我们需要创建一个具有单一端点的控制器。消息将发布到此端点，然后由我们的生产者处理。然后，我们的消费者将通过登录到控制台来捕获并处理它。

@RestController
@RequestMapping(value = <font>"/kafka"</font><font>)
<b>public</b> <b>class</b> KafkaController {

  <b>private</b> <b>final</b> Producer producer;

  @Autowired
  KafkaController(Producer producer) {
    <b>this</b>.producer = producer;
  }

  @PostMapping(value = </font><font>"/publish"</font><font>)
  <b>public</b> <b>void</b> sendMessageToKafkaTopic(@RequestParam(</font><font>"message"</font><font>) String message) {
    <b>this</b>.producer.sendMessage(message);
  }
}
</font>

让我们使用cURL将消息发送给Kafka：

curl -X POST -F 'message=test' http://localhost:9000/kafka/publish

基本上就是这样！在不到10个步骤中，您了解了将Apache Kafka添加到Spring Boot项目是多么容易。如果您遵循本指南，您现在知道如何将Kafka集成到Spring Boot项目中，并且您已准备好使用这个超级工具！

总结

以上所述是小编给大家介绍的在Spring Boot应用程序中使用Apache Kafka的方法步骤详解，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对脚本之家网站的支持！

您可能感兴趣的文章:

相关文章

  • 详谈spring中bean注入无效和new创建对象的区别

    详谈spring中bean注入无效和new创建对象的区别

    这篇文章主要介绍了spring中bean注入无效和new创建对象的区别，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2022-02-02
  • Mybatis-plus如何更新Null字段详解

    Mybatis-plus如何更新Null字段详解

    MyBatis-plus在进行更新操作时不会更新传入实体中为null或默认值属性字段,只更新不为null的值、非默认值的属性字段,这篇文章主要给大家介绍了关于Mybatis-plus如何更新Null字段的相关资料,需要的朋友可以参考下
    2023-07-07
  • SpringBoot中的文件上传和异常处理详解

    SpringBoot中的文件上传和异常处理详解

    这篇文章主要介绍了SpringBoot中的文件上传和异常处理详解,对于机器客户端,它将生成JSON响应,其中包含错误,HTTP状态和异常消息的详细信息,对于浏览器客户端,响应一个"whitelabel"错误视图,以HTML格式呈现相同的数据,需要的朋友可以参考下
    2023-09-09
  • Spark使用IDEA编写wordcount的示例演示

    Spark使用IDEA编写wordcount的示例演示

    这篇文章主要介绍了Spark使用IDEA编写wordcount的示例演示,本文通过示例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
    2023-07-07
  • 解决Mybatis中mapper.xml文件update,delete及insert返回值问题

    解决Mybatis中mapper.xml文件update,delete及insert返回值问题

    这篇文章主要介绍了解决Mybatis中mapper.xml文件update,delete及insert返回值问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2020-11-11
  • Seata 环境搭建部署过程

    Seata 环境搭建部署过程

    Seata是一个分布式事务，seata服务端也是一个微服务，需要和其他微服务一样需要注册中心和配置中心，这篇文章主要介绍了Seata 环境搭建,需要的朋友可以参考下
    2022-10-10
  • 聊聊mybatis sql的括号问题

    聊聊mybatis sql的括号问题

    这篇文章主要介绍了mybatis sql的括号问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2022-01-01
  • Java连接mysql数据库以及mysql驱动jar包下载和使用方法

    Java连接mysql数据库以及mysql驱动jar包下载和使用方法

    这篇文章主要给大家介绍了关于Java连接mysql数据库以及mysql驱动jar包下载和使用方法,MySQL是一款常用的关系型数据库,它的JDBC驱动程序使得我们可以通过Java程序连接MySQL数据库进行数据操作,需要的朋友可以参考下
    2023-11-11
  • Java @Validated遇到的大坑与处理

    Java @Validated遇到的大坑与处理

    这篇文章主要介绍了Java @Validated遇到的大坑与处理方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-12-12
  • springboot使用redisRepository和redistemplate操作redis的过程解析

    springboot使用redisRepository和redistemplate操作redis的过程解析

    本文给大家介绍springboot整合redis/分别用redisRepository和redistemplate操作redis，本文结合实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友参考下吧
    2022-05-05

最新评论