一、配置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文章来源:https://www.toymoban.com/news/detail-534283.html
文章来源地址https://www.toymoban.com/news/detail-534283.html
到了这里,关于如何轻松在SpringBoot中正确配置并运行Kafka的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!