【SpringBoot系列】SpringBoot整合Kafka(含源码)

本文介绍: 在现代的微服务架构中，消息队列已经成为了一个不可或缺的组件。它能够帮助我们在不同的服务之间传递消息，并且能够确保这些消息不会丢失。在众多的消息队列中，Kaf k a 是一个非常出色的选择。它能够处理大量的实时数据，并且提供了强大的持久化能力。在本文中，我们将会探讨如何在 Spr in gBo o t 中整合 Kaf k a。Apac he Kafka 是一个开源的流处理平台，由 Linke dIn 团队开发并于 2011 年贡献给 Apac he 基金会。Kafka 以其高吞吐量、可扩展性和容错性而闻名。

579a429daf314744b995f37351b46548

前言

在现代的微服务架构中，消息队列已经成为了一个不可或缺的组件。

它能够帮助我们在不同的服务之间传递消息，并且能够确保这些消息不会丢失。

在众多的消息队列中，Kafka 是一个非常出色的选择。

它能够处理大量的实时数据，并且提供了强大的持久化能力。

在本文中，我们将会探讨如何在 Spr in gBo o t 中整合 Kafka。

什么是Kafka?

Apac he Kafka 是一个开源的流处理平台，由 Linke dIn 团队开发并于 2011 年贡献给 Apac he 基金会。Kafka 以其高吞吐量、可扩展性和容错性而闻名。它是一个基于发布/订阅模式的消息系统，通常用于大型实时数据流处理应用。

Kafka 的主要组件包括：

Pro du ce r：负责发布消息到 Kafka 服务器。
Br ok er：是 Kafka 服务器实例，负责消息的存储、接收和发送。
Con sum er：从 Kafka 服务器读取消息。
Top ic：消息的类别或者说是消息的标签，Pro du cer 将消息发布到特定的 Top ic，Con sum er 从特定的 Top ic 读取消息。

Kafka 可以在分布式系统中用于构建实时流数据管道，它可以在系统或应用之间可靠地获取数据。此外，Kafka 可以和 Apac he St orm、Apac he Had oop、Apac he Spark 等进行集成，用于大数据处理和分析。

Kafka的应用场景?

日志 收集：

一个公司可能有很多服务器，每个服务器上运行着很多服务，Kafka 可以用来实现这些服务的日志收集功能。各服务的日志分别发送到 Kafka 的不同 Top ic 中。

消息系统：

Kafka 能够作为一个大规模的消息处理系统，各生产者将消息发送到 Kafka，消费者从 Kafka 中读取消息进行处理。

用户活动跟踪：

Kafka 也常用于用户活动跟踪和实时分析。例如，用户的点击、搜索等行为可以实时写入到 Kafka，然后进行实时或者离线分析。

在 Kafka 上可以进行实时的流处理。例如，使用 Apac he St orm 集成 Kafka 来进行实时的数据处理。

指标和日志聚合：

统计数据和监控数据也是 Kafka 的一个重要应用场景。例如，通过 Kafka 可以收集各种分布式应用的数据，然后进行统一的处理和分析。

事件源：

Kafka 可以作为大规模事件处理的源头，例如，用户的行为、系统的状态等都可以作为事件，通过 Kafka 进行分发处理。

示例

版本 依赖

模块	版本
Spr in gBo o t	3.1.0
JDK	17

代码

KafkaCo nf i g

@Configuration
@EnableKafka
public class KafkaConfig {

    @Bean
    public KafkaReceiver listener() {
        return new KafkaReceiver();
    }

}

KafkaSender

@Component
@Slf4j
public class KafkaSender {
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    public void send(String topic, String key, String data) {
        //发送消息
        CompletableFuture<SendResult<String, Object>> completable = kafkaTemplate.send(topic, key, data);
        completable.whenCompleteAsync((result, ex) -> {
            if (null == ex) {
                log.info(topic + "生产者发送消息成功：" + result.toString());
            } else {
                log.info(topic + "生产者发送消息失败：" + ex.getMessage());
            }
        });
    }
}

KafkaReceiver

@Component
@Slf4j
public class KafkaReceiver {
    /**
     * 下面的主题是一个数组，可以同时订阅多主题，只需按数组格式即可，也就是用","隔开
     */
    @KafkaListener(topics = {"testTopic"})
    public void receive(ConsumerRecord<?, ?> record){
        log.info("消费者收到的消息key: " + record.key());
        log.info("消费者收到的消息value: " + record.value().toString());
    }
}

KafkaCont roller

/**
 * kafka 测试接口
 */
@RestController
public class KafkaController {
    @Autowired
    private KafkaSender kafkaSender;

    @GetMapping("/sendMessageToKafka")
    public String sendMessageToKafka() {
        Map<String, String> messageMap = new HashMap();
        messageMap.put("message", "hello world!");
        ObjectMapper objectMapper = new ObjectMapper();
        String data = null;
        try {
            data = objectMapper.writeValueAsString(messageMap);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
        String key = String.valueOf(UUID.randomUUID());
        //kakfa的推送消息方法有多种，可以采取带有任务key的，也可以采取不带有的（不带时默认为null）
        kafkaSender.send("testTopic", key, data);
        return "ok";
    }
}