Spring Kafka 官方提供了一种基于消息类型的消费模式，通过类上的 @KafkaListener 注解配合方法上的 @KafkaHandler，实现多态消息的自动路由处理。其典型实现方式如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

@KafkaListener(id = "multi", topics = "myTopic")
static class MultiListenerBean {
    
    @KafkaHandler
    public void listen(String foo) {
        ...
    }
    
    @KafkaHandler
    public void listen(Integer bar) {
        ...
    }
    
    @KafkaHandler(isDefault = true)
    public void listenDefault(Object object) {
        ...
    }
}

根据 Spring Kafka 官方文档 @KafkaListener on a Class 的说明：

When messages are delivered, the converted message payload type is used to determine which method to call.
接收到消息后，会使用转换后的消息类型来决定调用哪个方法。

开发者可以基于此特性，轻松实现从简单文本消息到复杂领域事件的各种消息处理场景。

一、背景与现状

目前，我们团队在使用 Kafka 时主要存在以下几类问题：

• 问题一：一个工程对接多套 Kafka 集群，工程代码缺少对多套集群的规范管理；
• 问题二：生产环境分为基线和灰度，对接灰度 Kafka 时，工程改动繁琐；
• 问题三：生产/消费消息时，出现大量偏技术性的重复代码：序列化、反序列化、判空、重试等；
• 问题四：在发送消息、消费消息过程中，存在链路丢标的情况；

因此，建设统一消息组件，以解决上述问题，简化消息收发。

在 Java 生态中，Apache Kafka 通过 kafka-clients.jar 提供了原生客户端支持。开发者需要手动创建 KafkaConsumer 实例并订阅指定主题（Topic）来实现消息消费。典型实现如下：

1
2
3
4
5
6
7
8
9
10

public void pollMessages() {
    // 1. 初始化消费者实例
    Consumer<String, String> consumer = new KafkaConsumer<>(getConsumerConfig());
    // 2. 订阅主题并设置重平衡监听器
    consumer.subscribe(Collections.singleton(topic), new RebalanceListener());
    // 3. 轮询获取消息（超时时间1秒）
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    // 4. 同步提交偏移量
    consumer.commitSync();
}

Spring Kafka 在原生客户端基础上进行了深度封装，通过声明式注解显著简化了开发流程。例如，只需使用 @KafkaListener 注解即可实现消息监听：

1
2
3
4

@KafkaListener(id = "orderService", topics = "order.topic")
public void handleOrderEvent(ConsumerRecord<String, String> record) {
    // 业务处理逻辑
}

这种简洁的语法背后，Spring Kafka 实际上构建了一套完整的消费者（Consumer）管理机制。那么问题来了：Spring Kafka 是如何创建这些消费者的呢？

你说你 Kafka 用了很多年了，但是位移提交的这些细节你未必清楚。

不久前有个同事出去面试了，他说在 Kafka 问题上被面试官藐视了。

简介：传统的消息队列对业务方提出了更高的要求，我们期望提供的是一种以业务为重心的，面向服务的解决方案。

这里记录的是学习分享内容，文章维护在 Github：studeyang/leanrning-share。

我们在上次分享中聊到了领域驱动设计和微服务，在 DDD 中有一个术语叫做领域事件，例如订单模型中的订单已创建、商品已发货。领域事件会触发下一步的业务操作，如果领域事件发生在微服务内，可以通过观察者模式很容易实现消息监听并处理。

如果发生在微服务之间，则需引入事件总线或者消息中间件。