这里将生产者和消费者放在一个应用中
使用的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字符串
![[项目]基于FreeRTOS的STM32四轴飞行器: 八.遥控器摇杆](https://i-blog.csdnimg.cn/direct/ea23f4e4da9c400095e2af65a415ddbc.png)
