SpringBoot整合Kafka

这篇具有很好参考价值的文章主要介绍了SpringBoot整合Kafka。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、首先下载windows版本的Kafka

官网:Apache Kafka

二、启动Kafka

cmd进入到kafka安装目录:

1:cmd启动zookeeer

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

2:cmd启动kafka server

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

3:使用cmd窗口启动一个生产者命令:

.\bin\windows\kafka-console-producer.bat --bootstrap-server localhost:9092 --topic Topic1

4:cmd启动zookeeer

.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 -topic Topic1

 三、引入kafka依赖

       <!--kafka依赖-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

四、配置文件

server:
    port: 8080

spring:
    application:
        name: kafka-demo
    kafka:
        bootstrap-servers: localhost:9092
        producer: # producer 生产者
            retries: 0 # 重试次数
            acks: 1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
            batch-size: 16384 # 批量大小
            buffer-memory: 33554432 # 生产端缓冲区大小
            key-serializer: org.apache.kafka.common.serialization.StringSerializer
            #      value-serializer: com.itheima.demo.config.MySerializer
            value-serializer: org.apache.kafka.common.serialization.StringSerializer

        consumer: # consumer消费者
            group-id: javagroup # 默认的消费组ID
            enable-auto-commit: true # 是否自动提交offset
            auto-commit-interval: 100  # 提交offset延时(接收到消息后多久提交offset)

            # earliest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
            # latest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
            # none:topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常
            auto-offset-reset: earliest
            key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
            #      value-deserializer: com.itheima.demo.config.MyDeserializer
            value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

 五、编写生产者发送消息

1:异步发送

@RestController
@Api(tags = "异步接口")
@RequestMapping("/kafka")
public class KafkaProducer {
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    @GetMapping("/kafka/test/{msg}")
    public void sendMessage(@PathVariable("msg") String msg) {
        Message message = new Message();
        message.setMessage(msg);
        kafkaTemplate.send("Topic3", JSON.toJSONString(message));
    }
}

1:同步发送

//测试同步发送与监听
@RestController
@Api(tags = "同步接口")
@RequestMapping("/kafka")
public class AsyncProducer {
    private final static Logger logger = LoggerFactory.getLogger(AsyncProducer.class);
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    //同步发送
    @GetMapping("/kafka/sync/{msg}")
    public void sync(@PathVariable("msg") String msg) throws Exception {
        Message message = new Message();
        message.setMessage(msg);
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send("Topic3", JSON.toJSONString(message));
        //注意,可以设置等待时间,超出后,不再等候结果
        SendResult<String, Object> result = future.get(3, TimeUnit.SECONDS);
        logger.info("send result:{}",result.getProducerRecord().value());
    }

}

六、消费者编写

@Component
public class KafkaConsumer {
    private final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    //不指定group,默认取yml里配置的
    @KafkaListener(topics = {"Topic3"})
    public void onMessage1(ConsumerRecord<?, ?> consumerRecord) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            Object msg = optional.get();
            logger.info("message:{}", msg);
        }
    }
}

 通过swagger,进行生产者发送消息,观察控制台结果

SpringBoot整合Kafka,spring boot,kafka,后端

SpringBoot整合Kafka,spring boot,kafka,后端

 至此,一个简单的整合就完成了。

后续会持续更新kafka相关内容(多多关注哦!)

SpringBoot整合Kafka,spring boot,kafka,后端文章来源地址https://www.toymoban.com/news/detail-809695.html

到了这里,关于SpringBoot整合Kafka的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Spring Boot 整合kafka消费模式AckMode以及手动消费

    在pom.xml文件中导入依赖 需要自己配置AckMode时候的配置 kafka支持的消费模式,设置在 AbstractMessageListenerContainer.AckMode 的枚举中,下面就介绍下各个模式的区别 AckMode模式 AckMode模式 作用 MANUAL 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后, 手动调用Acknowledgment

    2024年02月08日
    浏览(43)
  • 第二章 Spring Boot 整合 Kafka消息队列 生产者

    第一章 Kafka 配置部署及SASL_PLAINTEXT安全认证 第二章  Spring Boot 整合 Kafka消息队列 生产者 第三章  Spring Boot 整合 Kafka消息队列 消息者         Kafka 是一个消息队列产品,基于Topic partitions的设计,能达到非常高的消息发送处理性能。本文主是基于Spirng Boot封装了Apache 的

    2024年01月25日
    浏览(46)
  • Spring Boot 整合kafka消费模式AckMode以及手动消费 依赖管理

    在pom.xml文件中导入依赖 需要自己配置AckMode时候的配置 kafka支持的消费模式,设置在 AbstractMessageListenerContainer.AckMode 的枚举中,下面就介绍下各个模式的区别 AckMode模式 AckMode模式 作用 MANUAL 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后, 手动调用Acknowledgment

    2024年02月16日
    浏览(45)
  • 从零到Kafka:万字带你体验Spring Boot整合消息驱动的奇妙之旅

    主页传送门:📀 传送 Spring boot : | 基于Spring的开源框架,用于简化新Spring应用的初始搭建以及开发过程 特性: | 快速开发、轻量级、无代码生成和独立运行等特性 优势: | 简化配置,提供自动配置,减少开发时间 应用场景: | 适用于微服务架构、云原生应用等场景 环境

    2024年02月05日
    浏览(43)
  • Spring Boot 整合kafka:生产者ack机制和消费者AckMode消费模式、手动提交ACK

    Kafka 生产者的 ACK 机制指的是生产者在发送消息后,对消息副本的确认机制。ACK 机制可以帮助生产者确保消息被成功写入 Kafka 集群中的多个副本,并在需要时获取确认信息。 Kafka 提供了三种 ACK 机制的配置选项,分别是: acks=0:生产者在成功将消息发送到网络缓冲区后即视

    2024年02月04日
    浏览(54)
  • Spring Boot 整合 Shiro(后端)

    1 Shiro 什么是 Shiro 官网: http://shiro.apache.org/ 是一款主流的 Java 安全框架,不依赖任何容器,可以运行在 Java SE 和 Java EE 项目中,它的主要作用是对访问系统的用户进行身份认证、 授权、会话管理、加密等操作。 Shiro 就是用来解决安全管理的系统化框架。 2 Shiro 核心组件 用

    2024年02月09日
    浏览(51)
  • java阻塞队列/kafka/spring整合kafka

    queue增加删除元素 增加元素 add方法在添加元素的时候,若超出了度列的长度会直接抛出异常: put方法,若向队尾添加元素的时候发现队列已经满了会发生阻塞一直等待空间,以加入元素 offer方法在添加元素时,如果发现队列已满无法添加的话,会直接返回false 删除元素 pol

    2024年02月12日
    浏览(46)
  • 什么是kafka,如何学习kafka,整合SpringBoot

    目录 一、什么是Kafka,如何学习 二、如何整合SpringBoot 三、Kafka的优势   Kafka是一种分布式的消息队列系统,它可以用于处理大量实时数据流 。学习Kafka需要掌握如何安装、配置和运行Kafka集群,以及如何使用Kafka API编写生产者和消费者代码来读写数据。此外,还需要了解Ka

    2024年02月10日
    浏览(40)
  • Spring整合kafka

    只用spring-kafka依赖就行  注入KafkaTemplate模板 消息发送  监听消息消费 测试发送 spring-kafka和kafka-clients结合使用(推荐)  消费者组件 生产者组件 生产消息和消费消息  注: 这里记录一下生产发生的问题 关于max.poll.interval.ms配置的问题,根据自己的业务配置poll拉去间隔等待

    2024年02月02日
    浏览(43)
  • SpringBoot整合Kafka

    官网:Apache Kafka cmd进入到kafka安装目录: 1:cmd启动zookeeer .binwindowszookeeper-server-start.bat .configzookeeper.properties 2:cmd启动kafka server .binwindowszookeeper-server-start.bat .configzookeeper.properties 3:使用cmd窗口启动一个生产者命令: .binwindowskafka-console-producer.bat --bootstrap-server local

    2024年01月20日
    浏览(38)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包