本文详解如何在 Spring Cloud Stream 3.1+ 中通过函数式编程模型（Functional Consumer）优雅处理多个 Kafka 主题，并基于消息头实现条件路由，避免废弃的 @StreamListener 和冗余的 MessageRoutingCallback 实现。
本文详解如何在 spring cloud stream 3.1+ 中通过函数式编程模型（functional consumer）优雅处理多个 kafka 主题，并基于消息头实现条件路由，避免废弃的 `@streamlistener` 和冗余的 `messageroutingcallback` 实现。
在 Spring Cloud Stream 进入函数式编程范式后，@StreamListener 已被正式弃用（自 3.1.0 起），取而代之的是以 java.util.function.Consumer、Function 和 Supplier 为核心的声明式绑定模型。面对“消费两个不同 Kafka 主题 + 第二主题需按 header 分支处理”的典型场景，关键不在于堆砌多个 MessageRoutingCallback，而在于合理划分职责边界：主题级路由由 binding 配置驱动，消息级条件路由交由 SpEL 表达式或单点 MessageRoutingCallback 承担。
✅ 正确架构：Binding 驱动主题分离 + SpEL 驱动 Header 分支
首先，明确一个核心原则：每个 Kafka 主题应绑定到独立的函数 bean（如 Consumer），而非共用一个 router bean。你遇到的启动失败（Parameter 3 of method functionRouter ... 2 were found）正是因为 Spring Cloud Function 的自动配置期望全局唯一 MessageRoutingCallback —— 但你并不需要它来分发主题，binding 本身已承担该职责。
✅ 方案一：纯配置化 SpEL 路由（推荐，简洁清晰）
适用于第二主题中 header 分支逻辑较简单（如根据 type=ORDER 或 type=PAYMENT 调用不同处理器）。无需编写任何 MessageRoutingCallback 类：
# application.yml
spring:
cloud:
stream:
# 定义两个独立函数：分别处理 topic-1-input 和 topic-2-input
function:
definition: firstConsumer;secondConsumerRouter
bindings:
# 主题1 → 直连 firstConsumer（无分支）
firstConsumer-in-0:
destination: topic-1-input
group: group-3
# 主题2 → 先进 secondConsumerRouter（带 SpEL 路由）
secondConsumerRouter-in-0:
destination: topic-2-input
group: group-4
# 关键：为 secondConsumerRouter 启用基于 header 的 SpEL 路由
rabbit: {} # 若用 RabbitMQ 可忽略；Kafka 用户需确保版本 ≥ 3.4.x（原生支持 SpEL routing）
kafka:
binder:
configuration:
default:
key:
serializer: org.apache.kafka.common.serialization.StringSerializer
value:
serializer: org.apache.kafka.common.serialization.StringSerializer// Java 配置：定义两个函数 Bean
@Configuration
@Slf4j
public class StreamFunctionConfig {
// 处理 topic-1-input：统一逻辑
@Bean
public Consumer<Message<String>> firstConsumer() {
return message -> {
log.info("✅ Topic-1 | Payload: {}, Headers: {}",
message.getPayload(), message.getHeaders());
// 执行统一业务逻辑
};
}
// 处理 topic-2-input：作为路由入口，委托给子函数
@Bean
public Function<Message<String>, Message<?>> secondConsumerRouter() {
return message -> {
String type = (String) message.getHeaders().get("type");
log.info("? Topic-2 routing by header 'type' = {}", type);
// 根据 header 构造新消息并路由到对应函数（模拟路由行为）
// 注意：实际中建议使用 SpEL（见下方 YAML 替代方案），此处为演示逻辑
if ("ORDER".equalsIgnoreCase(type)) {
return MessageBuilder.fromMessage(message)
.setHeader("spring.cloud.stream.function.definition", "orderHandler")
.build();
} else if ("PAYMENT".equalsIgnoreCase(type)) {
return MessageBuilder.fromMessage(message)
.setHeader("spring.cloud.stream.function.definition", "paymentHandler")
.build();
}
throw new IllegalArgumentException("Unknown type: " + type);
};
}
// 子处理器：Order 专用逻辑
@Bean
public Consumer<Message<String>> orderHandler() {
return message -> {
log.info("? ORDER handler | Payload: {}", message.getPayload());
// 订单专属处理
};
}
// 子处理器：Payment 专用逻辑
@Bean
public Consumer<Message<String>> paymentHandler() {
return message -> {
log.info("? PAYMENT handler | Payload: {}", message.getPayload());
// 支付专属处理
};
}
}⚠️ 注意：上述 Java 路由是逻辑示意。生产环境强烈推荐使用 SpEL 表达式路由（更轻量、解耦、可配置化），需配合 Spring Cloud Stream 3.4+ 和 Kafka Binder：
# application.yml（SpEL 路由版，无需 Java Router）
spring:
cloud:
stream:
function:
definition: firstConsumer;orderHandler;paymentHandler
bindings:
firstConsumer-in-0:
destination: topic-1-input
group: group-3
# 绑定 topic-2-input 到路由函数（名称需匹配）
router-in-0:
destination: topic-2-input
group: group-4
# ? 关键：启用 SpEL 路由规则
router:
expression: headers['type'] == 'ORDER' ? 'orderHandler' : headers['type'] == 'PAYMENT' ? 'paymentHandler' : ''此时只需定义 orderHandler 和 paymentHandler 两个 @Bean Consumer，无需任何 MessageRoutingCallback —— 框架自动根据 headers.type 将消息路由到对应函数。
✅ 方案二：单点 MessageRoutingCallback（复杂逻辑时）
若 header 解析逻辑涉及服务调用、数据库查询等，SpEL 不足以胜任，则仅定义一个 MessageRoutingCallback，并在其中集中处理所有路由决策：
@Component
@Slf4j
public class UnifiedMessageRouter implements MessageRoutingCallback {
private final OrderService orderService;
private final PaymentService paymentService;
public UnifiedMessageRouter(OrderService orderService, PaymentService paymentService) {
this.orderService = orderService;
this.paymentService = paymentService;
}
@Override
public FunctionRoutingResult routingResult(Message<?> message) {
String type = (String) message.getHeaders().get("type");
String topic = (String) message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC);
if ("topic-1-input".equals(topic)) {
// 主题1不参与路由，直连 firstConsumer（binding 已保证）
return new FunctionRoutingResult("firstConsumer");
}
if ("topic-2-input".equals(topic)) {
switch (type) {
case "ORDER":
return new FunctionRoutingResult("orderHandler");
case "PAYMENT":
return new FunctionRoutingResult("paymentHandler");
case "NOTIFICATION":
// 复杂逻辑示例：动态查库决定处理器
String handlerName = paymentService.resolveNotificationHandler(
(String) message.getPayload()
);
return new FunctionRoutingResult(handlerName);
default:
log.warn("No route for type: {}", type);
return null; // 跳过此消息
}
}
return null;
}
}同时更新配置，指向该唯一 Router：
spring:
cloud:
stream:
function:
definition: firstConsumer;orderHandler;paymentHandler
bindings:
firstConsumer-in-0:
destination: topic-1-input
group: group-3
# 绑定 topic-2-input 到 router（router 会再分发）
router-in-0:
destination: topic-2-input
group: group-4? 总结与最佳实践
- ❌ 不要为每个主题创建独立 MessageRoutingCallback —— 违反设计初衷，触发框架冲突；
- ✅ 优先使用 SpEL 表达式路由（spring.cloud.stream.router.expression）处理 header 分支，零代码、高可维护；
- ✅ 复杂路由逻辑才引入单点 MessageRoutingCallback，保持其为应用内唯一路由中枢；
- ✅ 主题隔离靠 binding 配置，而非 router 逻辑 —— firstConsumer-in-0 和 secondConsumer-in-0 应直接绑定不同 topic；
- ✅ 函数名（如 firstConsumer）必须与 spring.cloud.stream.function.definition 中定义的名称严格一致；
- ✅ 确保 Kafka Binder 版本兼容（推荐 Spring Cloud 2022.0.x / Spring Boot 3.0+ + Spring Cloud Stream Horsham SR12+）。
通过以上方式，你将获得清晰、可测、易扩展的函数式消息处理架构，彻底告别 @StreamListener 和混乱的多 Router 抗模式。
