【Spring Boot实战与进阶】集成Kafka消息队列

程序员云帆哥

发布于 2022-12-27 16:58:01

7210

发布于 2022-12-27 16:58:01

文章被收录于专栏：程序员云帆哥程序员云帆哥

汇总目录链接：【Spring Boot实战与进阶】学习目录

文章目录
一、简介
二、集成Kafka消息队列
1、引入依赖
2、配置文件
3、测试生产消息
4、测试消费消息

一、简介

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

二、集成Kafka消息队列

1、引入依赖

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

2、配置文件

spring:
  kafka:
    bootstrap-servers: 192.168.60.123:9092
    producer:
      retries: 0
    consumer:
      group-id: test

3、测试生产消息

@SpringBootTest
class ScriptApplicationTests {
    @Resource
    private KafkaTemplate kafkaTemplate;

    @Test
    void contextLoads() {
        kafkaTemplate.send("delay-message", "123456");
    }
}

看下Kafka的UI界面（Kafka-map）：

4、测试消费消息

@Slf4j
@Component
public class KafkaConsumer {

    @KafkaListener(topics = "delay-message")
    public void ProductInsertEvent1(ConsumerRecord<String, String> record) {
        Optional<String> kafkaMessage = Optional.ofNullable(record.value());
        kafkaMessage.ifPresent(msg -> {
            log.info("KafkaConsumer kafka value:{}", msg);
        });
    }

}