本篇文章给大家分享的是有关如何在SpringBoot项目中使用Kafka,小编觉得挺实用的,因此分享给大家学习,希望大家阅读完这篇文章后可以有所收获,话不多说,跟着小编一起来看看吧。
搭建kafka 和 zookeeper 环境 并且启动 它们
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</dependency>
spring:
kafka:
bootstrap-servers: 192.168.25.6:9092 #bootstrap-servers:连接kafka的地址,多个地址用逗号分隔
consumer:
group-id: myGroup
enable-auto-commit: true
auto-commit-interval: 100ms
properties:
session.timeout.ms: 15000
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
auto-offset-reset: earliest
producer:
retries: 0 #若设置大于0的值,客户端会将发送失败的记录重新发送
batch-size: 16384 #当将多个记录被发送到同一个分区时, Producer 将尝试将记录组合到更少的请求中。这有助于提升客户端和服务器端的性能。这个配置控制一个批次的默认大小(以字节为单位)。16384是缺省的配置
buffer-memory: 33554432 #Producer 用来缓冲等待被发送到服务器的记录的总字节数,33554432是缺省配置
key-serializer: org.apache.kafka.common.serialization.StringSerializer #关键字的序列化类
value-serializer: org.apache.kafka.common.serialization.StringSerializer #值的序列化类
/**
* @author johnny
* @create 2020-09-23 上午9:21
**/
@Data
public class Message {
private Long id;
private String msg;
private Date sendTime;
}
主要利用 KafkaTemplate 来发送消息 ,将消息封装成Message 并且进行 转化成Json串 发送到Kafka中
@Component
@Slf4j
public class KafkaSender {
private final KafkaTemplate<String, String> kafkaTemplate;
//构造器方式注入 kafkaTemplate
public KafkaSender(KafkaTemplate<String, String> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
private Gson gson = new GsonBuilder().create();
public void send(String msg) {
Message message = new Message();
message.setId(System.currentTimeMillis());
message.setMsg(msg);
message.setSendTime(new Date());
log.info("【++++++++++++++++++ message :{}】", gson.toJson(message));
//对 topic = hello2 的发送消息
kafkaTemplate.send("hello2",gson.toJson(message));
}
}
在监听的方法上通过注解配置一个监听器即可,另外就是指定需要监听的topic
kafka的消息再接收端会被封装成ConsumerRecord对象返回,它内部的value属性就是实际的消息。
@Component
@Slf4j
public class KafkaConsumer {
@KafkaListener(topics = {"hello2"})
public void listen(ConsumerRecord<?, ?> record) {
Optional.ofNullable(record.value())
.ifPresent(message -> {
log.info("【+++++++++++++++++ record = {} 】", record);
log.info("【+++++++++++++++++ message = {}】", message);
});
}
}
提供一个 Http接口调用 KafkaSender 去发送消息
@RestController
@Slf4j
public class TestController {
@Autowired
private KafkaSender kafkaSender;
@GetMapping("sendMessage/{msg}")
public void sendMessage(@PathVariable("msg") String msg){
kafkaSender.send(msg);
}
}
监听8080 端口
KafkaMessageListenerContainer中有 consumer group = myGroup 有一个 监听 hello2-0 topic 的 消费者
http://localhost:8080/sendMessage/KafkaTestMsg
以上就是如何在SpringBoot项目中使用Kafka,小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注亿速云行业资讯频道。
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。