SpringBoot如何正确配置并运行Kafka

 更新时间:2023年07月19日 09:55:32   作者:学弟不想努力了  
这篇文章主要介绍了SpringBoot如何正确配置并运行Kafka问题,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教

一、配置pom.xml,引入maven依赖

<!-- 引入kafka依赖 -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.8.6</version>
</dependency>

二、application.yml配置文件

这里只提供了kafka有用的相关配置,其他的配置删了

spring:
  kafka:
    bootstrap-servers: xx.xx.xx.xx:9092 # kafka集群信息,多个用逗号间隔
    # 生产者
    producer:
      # 重试次数,设置大于0的值,则客户端会将发送失败的记录重新发送
      retries: 3
      batch-size: 16384 #批量处理大小,16K
      buffer-memory: 33554432 #缓冲存储大,32M
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    # 消费者
    consumer:
      # 消费者组
      group-id: TestGroup
      # 是否自动提交
      enable-auto-commit: false
      # 消费偏移配置
      # none:如果没有为消费者找到先前的offset的值,即没有自动维护偏移量,也没有手动维护偏移量,则抛出异常
      # earliest:在各分区下有提交的offset时:从offset处开始消费;在各分区下无提交的offset时:从头开始消费
      # latest:在各分区下有提交的offset时:从offset处开始消费;在各分区下无提交的offset时:从最新的数据开始消费
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    # 监听
    listener:
      # record:当每一条记录被消费者监听器(ListenerConsumer)处理之后提交
      # batch:当每一批poll()的数据被ListenerConsumer处理之后提交
      # time:当每一批poll()的数据被ListenerConsumer处理之后,距离上次提交时间大于TIME时提交
      # count:当每一批poll()的数据被ListenerConsumer处理之后,被处理record数量大于等于COUNT时提交
      # count_time:TIME或COUNT中有一个条件满足时提交
      # manual:当每一批poll()的数据被ListenerConsumer处理之后, 手动调用Acknowledgment.acknowledge()后提交
      # manual_immediate:手动调用Acknowledgment.acknowledge()后立即提交,一般推荐使用这种
      ack-mode: manual_immediate

三、消费者

消费者监听,可以配置多个监听器

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
/**
 * 消费者
 * kafka监听器
 */
@Component
public class KafkaConsumer {
    /**
     * kafka的监听器1,topic为"topic_test",消费者组为"group_topic_test"
     * @param record
     * @param item
     */
    @KafkaListener(topics = "topic_test", groupId = "group_topic_test")
    public void topicListener1(ConsumerRecord<String, String> record, Acknowledgment item) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手动提交
        item.acknowledge();
    }
    /**
     * 配置多个消费组
     * kafka的监听器2,topic为"topic_test2",消费者组为"group_topic_test"
     * @param record
     * @param item
     */
    @KafkaListener(topics = "topic_test2",groupId = "group_topic_test2")
    public void topicListener2(ConsumerRecord<String, String> record, Acknowledgment item) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        item.acknowledge();
    }
}

四、生产者

生产者作为接口Api作为测试

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
 * kafka生产者
 */
@RestController
@RequestMapping("/kafka")
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @RequestMapping("/send")
    public void send() {
        kafkaTemplate.send("topic_test",  "key", "测试kafka消息");
    }
}

五、调用测试

启动Boot项目,使用Postman工具发送GET请求:

http://localhost:8080/kafka/send

 

总结

以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。

相关文章

  • Spring Boot 中starter的原理详析

    Spring Boot 中starter的原理详析

    这篇文章主要介绍了Spring Boot 中starter原理详析,文章围绕主题展开springboot的插件机制和starter原理相关资料,需要的小伙伴可以参考一下
    2022-06-06
  • Java利用Phantomjs实现生成图片的功能

    Java利用Phantomjs实现生成图片的功能

    这篇文章主要介绍了Java利用Phantomjs实现生成图片的功能,文中讲解非常细致,代码帮助大家更好的理解和学习,感兴趣的朋友可以了解下
    2020-08-08
  • Java实现将图片上传到webapp路径下 路径获取方式

    Java实现将图片上传到webapp路径下 路径获取方式

    这篇文章主要介绍了Java实现将图片上传到webapp路径下 路径获取方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-11-11
  • SpringBoot ApplicationContext接口深入分析

    SpringBoot ApplicationContext接口深入分析

    ApplicationContext是Spring应用程序中的中央接口,由于继承了多个组件,使得ApplicationContext拥有了许多Spring的核心功能,如获取bean组件,注册监听事件,加载资源文件等
    2022-11-11
  • Java如何利用Socket进行数据读写

    Java如何利用Socket进行数据读写

    这篇文章主要介绍了Java如何利用Socket进行数据读写,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-10-10
  • java 基础教程之多线程详解及简单实例

    java 基础教程之多线程详解及简单实例

    这篇文章主要介绍了java 基础教程之多线程详解及简单实例的相关资料,线程的基本属性、如何创建线程、线程的状态切换以及线程通信,需要的朋友可以参考下
    2017-03-03
  • MAC下如何设置JDK环境变量

    MAC下如何设置JDK环境变量

    这篇文章主要介绍了MAC下如何设置JDK环境变量问题,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2022-11-11
  • springboot整合Mybatis、JPA、Redis的示例代码

    springboot整合Mybatis、JPA、Redis的示例代码

    这篇文章主要介绍了springboot整合Mybatis、JPA、Redis的示例代码,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2020-09-09
  • java 伪造http请求ip地址的方法

    java 伪造http请求ip地址的方法

    这篇文章主要介绍了java 伪造http请求ip地址的方法,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2018-09-09
  • javaweb项目入门实战详细指南

    javaweb项目入门实战详细指南

    Java Web开发是指使用Java技术来创建动态网站或Web应用程序,Java Web开发主要使用Servlet、JSP(JavaServer Pages)、JavaBeans等技术来实现动态页面和处理业务逻辑,这篇文章主要介绍了javaweb项目入门的相关资料,需要的朋友可以参考下
    2025-12-12

最新评论