当前位置： 首页 > news >正文

SpringBoot基础Kafka示例

news 2025/11/4 19:26:03

这里将生产者和消费者放在一个应用中

使用的Boot3.4.3

引入Kafka依赖

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

yml配置


spring:
  application:
    name: kafka-1

  #kafka连接地址
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    #配置生产者
    producer:
      #消息发送失败重试次数
      retries: 0
      #一个批次可以使用内存的大小
      batch-size: 16384
      #一个批次消息数量
      buffer-memory: 33554432
      #键的序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      #值的序列化方式
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all


    consumer:
      #是否自动提交
      enable-auto-commit: false
      #自动提交的频率
      auto-commit-interval: 1000
      #earliest	从分区的最早偏移量开始消费	需要消费所有历史消息  latest	从分区的最新偏移量开始消费，忽略历史消息	只关心新消息
      #none	如果没有有效的偏移量，抛出异常	严格要求偏移量必须存在
      #exception spring-kafka不支持
      auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

    listener:
      #用于配置消费者如何处理消息的确认  ack配置方式  这里指定由消费者手动提交偏移量
      #Acknowledgment.acknowledge() 方法来提交偏移量
      ack-mode: MANUAL_IMMEDIATE
      concurrency: 4
test-1: group-1
test-2: group-2
test-3: group-3

server:
  port: 8099

生产者示例,一般可能是一个MQTT接收消息入口

package com.hrui.kafka1.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author hrui
 * @date 2025/3/10 14:56
 */
@RestController
public class EventProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping("/sendMessage")
    public String sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
        return "Message sent to topic '" + topic + "': " + message;
    }


    @RequestMapping("/sendMessage2")
    public String sendMessage2() {
        //通过构建器模式创建Message对象
        Message<String> message = MessageBuilder.withPayload("Hello, Kafka!")
                .setHeader(KafkaHeaders.TOPIC, "ceshi")
                .build();
        kafkaTemplate.send(message);
        return "Message sent to topic";
    }

}

消费者示例

注意:如果配置了手动提交ack,那么

主要目的不仅仅是避免重复消费，而是为了确保消息的可靠处理和偏移量（offset）的正确提交。它可以避免重复消费，但更重要的是保证消息不会丢失，并且在消息处理失败时能够重新消费。

package com.hrui.kafka1.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.jms.AcknowledgeMode;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

/**
 * @author hrui
 * @date 2025/3/10 15:57
 */
@Component
public class EventConsumer {

    @KafkaListener(topics = {"ceshi"},groupId = "#{'${test-1}'}")
    public void onMessage(ConsumerRecord<String,String> message){
        System.out.println("接收到消息1："+message.value());
    }

    @KafkaListener(topics = {"ceshi"},groupId = "#{'${test-2}'}")
    public void onMessage(String message){
        System.out.println("接收到消息2："+message);
    }

    @KafkaListener(topics = {"ceshi"}, groupId = "#{'${test-3}'}")
    public void onMessage(ConsumerRecord<String, String> message, Acknowledgment ack,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                          @Header(KafkaHeaders.GROUP_ID) String groupId) {
        try {
            System.out.println("接收到消息3：" + message + ", ack:" + ack + ", topic:" + topic + ", groupId:" + groupId);
            // 处理消息逻辑
            // ...


        } catch (Exception e) {
            // 处理异常，记录日志
            System.err.println("处理消息失败: " + e.getMessage());
            // 可以根据业务需求决定是否重新抛出异常
        }finally {
            // 手动提交偏移量
            ack.acknowledge();
        }
    }
}

生产者可选择异步或者同步发送消息

生产者发送消息有同步异步之说 那么消费者在消费消息时候 有没有同步异步之说呢???

在 Kafka 消费者中，消费消息的方式本质上是由 Kafka 的设计决定的，而不是由消费者代码显式控制的。Kafka 消费者在消费消息时，通常是以拉取（poll）的方式从 Kafka 服务器获取消息，然后处理这些消息。从这个角度来看，消费者的消费行为是同步的，因为消费者需要主动调用 poll 方法来获取消息。

然而，消费者的消息处理逻辑可以是同步或异步的，具体取决于业务实现。以下是对消费者消费消息的同步和异步行为的详细分析：

 消费者的同步消费

在默认情况下，Kafka 消费者的消费行为是同步的，即：

  • 消费者通过 poll 方法从 Kafka 拉取一批消息。

  • 消费者逐条处理这些消息。

  • 每条消息处理完成后，消费者提交偏移量（offset）。

  • 消费者继续调用 poll 方法获取下一批消息。

特点：

  • 消息处理是顺序的，即一条消息处理完成后才会处理下一条消息。

  • 如果某条消息处理时间较长，会影响后续消息的处理速度。

  • 适合消息处理逻辑简单、处理时间较短的场景。

@KafkaListener(topics = {"ceshi"}, groupId = "#{'${test-3}'}")
public void onMessage(ConsumerRecord<String, String> message, Acknowledgment ack) {
    try {
        System.out.println("接收到消息：" + message.value());
        // 同步处理消息逻辑
        processMessage(message);
    } catch (Exception e) {
        System.err.println("处理消息失败: " + e.getMessage());
    } finally {
        ack.acknowledge(); // 手动提交偏移量
    }
}

private void processMessage(ConsumerRecord<String, String> message) {
    // 模拟消息处理逻辑
    try {
        Thread.sleep(1000); // 假设处理一条消息需要 1 秒
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

2. 消费者的异步消费

在某些场景下，消费者可能需要以异步的方式处理消息，即：

  • 消费者通过 poll 方法拉取一批消息。

  • 将每条消息提交到一个线程池或异步任务中处理。

  • 消费者继续调用 poll 方法获取下一批消息，而不等待上一条消息处理完成。

特点：

  • 消息处理是并发的，可以提高消息处理的吞吐量。

  • 需要额外的线程池或异步任务管理机制。

  • 适合消息处理逻辑复杂、处理时间较长的场景。

示例代码：

@Autowired
private ExecutorService executorService; // 注入线程池

@KafkaListener(topics = {"ceshi"}, groupId = "#{'${test-3}'}")
public void onMessage(ConsumerRecord<String, String> message, Acknowledgment ack) {
    if (!StringUtils.hasText(message.value())) {
        ack.acknowledge();
        return;
    }

    // 提交异步任务处理消息
    executorService.submit(() -> {
        try {
            System.out.println("接收到消息：" + message.value());
            processMessage(message); // 异步处理消息
        } catch (Exception e) {
            System.err.println("处理消息失败: " + e.getMessage());
        } finally {
            ack.acknowledge(); // 手动提交偏移量
        }
    });
}

private void processMessage(ConsumerRecord<String, String> message) {
    // 模拟消息处理逻辑
    try {
        Thread.sleep(1000); // 假设处理一条消息需要 1 秒
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

同步代码示例

@RequestMapping("/sendMessage2")
    public String sendMessage2(){
        //通过构建器模式创建Message对象
        Message<String> message = MessageBuilder.withPayload("Hello, Kafka!")
                .setHeader(KafkaHeaders.TOPIC, "ceshi")
                .build();

        CompletableFuture<SendResult<String, String>> send = kafkaTemplate.send(message);
        try {
            //阻塞等待拿结果
            SendResult<String, String> sendResult = send.get();
            System.out.println("说明消息发送成功,如果不成功会抛出异常");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        return "Message sent to topic";
    }

异步注册回调的方式

 @RequestMapping("/sendMessage2")
    public String sendMessage2(){
        //通过构建器模式创建Message对象
        Message<String> message = MessageBuilder.withPayload("Hello, Kafka!")
                .setHeader(KafkaHeaders.TOPIC, "ceshi")
                .build();

        CompletableFuture<SendResult<String, String>> send = kafkaTemplate.send(message);
        //非阻塞  异步 注册回调异步通知
        send.thenAccept(result -> {
            System.out.println("消息发送成功");
        }).exceptionally(e->{
            System.out.println("发送失败");
            e.printStackTrace();
            return null;
        });


        return "Message sent to topic";
    }

如果需要发送的不是String类型 

那么要发送的不是String类型

KafkaTemplate<String,Object> kafkaTemplate;

一般来说可以专成JSON字符串发送

在引入spring-kafka的时候     KafkaAutoConfiguration中  配置了KafkaTemplate

Kafka<Object,Object>

如果需要用KafkaTemplate发送对象的时候

默认用的String序列化   会报错   除非将对象转为JSON字符串(一般可以这么做)

如果用对象的话   改成JsonSerializer  这样自动转JSON字符串

http://www.dtcms.com/a/62770.html

相关文章：

  • 深度学习subword分词BPE
  • 在 DEM 中模拟粒子破损
  • 领域驱动设计（DDD）是什么？
  • 深度剖析Redis：双写一致性问题及解决方案全景解析
  • Vue3实战学习（Vue3集成Vue-Router(路由跳转、编程式路由跳转。路由跳转的单参数、多参数传递。设置默认页面路由)）(上)(7)
  • 位运算的应用3（获取⼆进制中的指定位、leetcode 190.颠倒二进制位）
  • Word 小黑第15套
  • Linux_16进程地址空间
  • 音视频软件工程师面试题
  • JDK安装过程中误删path怎么办？
  • 开发、科研、日常办公工具汇总（自用，持续更新）
  • 一个基于LSTM的字符级文本生成模型的训练+使用(pytorch)
  • 【threejs实战教程一】初识Three.js，场景Scene、相机Camera、渲染器Renderer
  • mysql索引机制深度剖析
  • SVT-AV1源码分析函数 svt_av1_optimize_b
  • react中字段响应式
  • 简述你对 Spring MVC 的理解
  • GRU门控循环单元
  • android用java设置button之间的间距 笔记250311
  • 高效微调算法 (Parameter-Efficient Fine-tuning, PEFT) 详解
  • 深度学习与大模型-张量
  • 一键换肤的Qt-Advanced-Stylesheets
  • Linux账号和权限管理
  • 【Spring】AOP在实际项目中的运用
  • 程序化广告行业（6/89）：现状、未来与核心要点剖析
  • 使用Process Explorer、Dependency Walker和PE信息查看工具快速排查dll动态库因库与库版本不一致导致的加载失败问题
  • 支持selenium的chrome driver更新到134.0.6998.88
  • SpringBoot开发——整合SpringReport开源报表工具
  • 如何在 React 中使用 CSS-in-JS？
  • 【Go每日一练】构建一个简单的用户信息管理系统