Stream、Function、Task 与 Contract：框架抽象与集成边界
业务代码写的是 Function<String, String>、Consumer<Order> 和 StreamBridge.send("orders-out-0", order)；这些接口既不认识 Kafka 的日志、复制与位点，也不认识 RabbitMQ 的 exchange 与 delivery tag。Spring Cloud Stream 的价值是把这个差异装进 binder SPI，代价是“发送成功”“消费确认”“提交位点”“进入 DLQ”这些词在框架层与 broker 层含义不同。同一票里的 Function、Task、Contract 各有各的边界：FunctionCatalog 管函数注册与组合，不保证消息路由；Task 记录一次短生命周期执行的退出状态，不负责触发与恢复；Contract 生成消费者驱动契约测试，不证明生产兼容。本页按 25 个知识点拆开这四层，并在每一处标明谁提供抽象、谁提供保证。
先消歧一个同名概念：Feign 的 Contract 是“把接口注解解析成请求模板”的 SPI（远程客户端与容错组合：Feign 与 Resilience4j 的 Contract 与首个请求 一节），Spring Cloud Contract 是消费者驱动契约测试的工具链。“契约”一词在本页指后者，与 Feign 的 SpringMvcContract 没有代码关系。
本页对应 U36 / #73、PRD L05。Kafka 的复制、位点、重平衡与重试语义权威在 MQ生产存储消费可靠性、Kafka消费位点与重平衡、重试、死信、延迟消息与消息积压（U57/U58）；Spring Kafka/AMQP 直连归 U48；Batch/Quartz 归 U46；幂等与 Outbox 归 U61。本页只讲框架如何映射到这些语义，不重写 broker 内部机制。
版本、模块与边界
查询日期为 2026-10-28；移交公告正文沿用2026-10-27快照，本次访问公告返回403，项目API、归档状态及BOM已重新核对。工程基线 JDK 21、Maven 3.9.9、Boot 4.0.9、Cloud 2025.1.4。Cloud 2025.1.4 BOM 管理 Stream 5.0.4、Function 5.0.5、Task 5.0.2、Bus 5.0.4；Boot 管理 Spring Kafka / spring-kafka-test 4.0.8、Kafka 4.1.2、H2 2.4.240。Contract 5.0.3 不在 2025.1.4 BOM 中：Contract 曾随 2025.1.0–2025.1.2 列车（5.0.0–5.0.3），自 2025.1.3 起从 BOM 移除（spring-cloud-release 提交 9f2fd5e7131a713cee8a9a9868ca8590b2d0ffff “Removing spring-cloud-contract”，2026-07-06），使用时必须显式钉版本。
| 子项目 | 版本 | 固定提交（源码核对与摘录） | 维护状态 |
|---|---|---|---|
| spring-cloud-stream | 5.0.4 | c7532f9c9dfc46622cd640efc846a1911f308582 | 随 2025.1.x 列车维护 |
| spring-cloud-function | 5.0.5 | aca2beca6a370e9d667cc8ac42824470253427cd | 随 2025.1.x 列车维护 |
| spring-cloud-task | 5.0.2 | 6e889bc68a7057577db32c61eb844b65773a33b1 | 随 2025.1.x 列车维护 |
| spring-cloud-contract | 5.0.3（显式钉版本） | 5d0aacd4c0651b1b330fc2ed5406729586301583 | EOL：官方已停止维护 |
| spring-cloud-release（BOM） | 2025.1.4 | 5bf63ab32b35f894129e0dc8602f4a20ddd2e3fa | 列车选择依据 |
Contract 的维护状态不是本页推断：api.spring.io 项目状态为 END_OF_LIFE；GitHub 仓库已 301 迁至 spring-attic/spring-cloud-contract 且 archived=true；2026-07-06 官方宣布移交 Stubborn.sh、从所有未来 release train 移除；5.0.3 是最后一个 patch（Maven Central 发布于 2026-06-11，父 POM 基于 Boot 4.0.7，其 pom.xml:26 固定 spring-boot.version=4.0.7）。因此 Contract 章节在本页的定位是固定版本 + 维护状态/迁移案例，不是“在维护项目的推荐用法”。缓存的 v4.3.0 快照（1e8a89ab…，基于 Boot 3.5）与运行版本错配，本页不引用。
五个固定提交分别对应 Stream、Function、Task、Contract 与 BOM 列车；source-excerpt 的 tag 字段使用完整提交，SHA-256 对整文件计算，代码为连续原文；教学代码另行标注。
| 模块 | 教学落点与边界 |
|---|---|
| spring-cloud-stream / Kafka、Kafka Streams、Pulsar binder / schema registry | 命令式绑定、拓扑与状态、订阅/schema、版本登记分别讲；代表Kafka路径实跑，其余推演 |
| stream-binder-rabbit | Rabbit 映射的配置面与源码；本机无 RabbitMQ，未运行 |
| spring-cloud-function core/context/web 与 adapters/integration/Kotlin | Catalog/组合/路由/Flux实跑；HTTP、平台/协议与流程接点推演；Kotlin内部模块不作生产starter |
| spring-cloud-task core/batch/stream/single-step starter | 执行记录与单实例实跑；Batch关联、单步装配和生命周期消息桥推演，不等于调度器 |
| spring-cloud-contract-verifier / -stub-runner | DSL、生成测试、stub 运行；EOL，按固定 5.0.3 讲解 |
| starter 家族 | 依赖入口，不证明 broker 可用；自动配置只证明装配 |
固定根POM递归核查了111个活动POM（含父聚合、BOM、docs与上游测试），不把它们当成111个课程点。Kafka Streams/Pulsar/schema registry、Function的各adapter/integration/Kotlin、Task Batch/Stream/单步starter及Contract三种DSL/插件/服务封装均有去向；聚合、文档与样例按所支撑能力登记，完整云厂商资源/IAM/计费交付沿用PRD的X边界。逐POM来源与处置保存在仓库 docs/audits/java-depth-u36-modules.json；旧reactive Kafka与deployer文档冲突在对应节解释。
本页与相邻票的入口区分：Stream 是 binder 抽象（配置键 spring.cloud.stream.*）；Spring Kafka/AMQP 直连（U48）是 spring.kafka.*/spring.rabbitmq.* 下的容器与模板；broker 语义（U57/U58）是日志、复制、位点与确认协议。Boot 的“消息四种接线”见Boot集成：自动配置条件与替换 的 消息：四种接线和确认边界，那只证明 starter 装配，不证明 Stream binding 生命周期。RabbitMQ 本机不可用（AGENTS 列出的本机服务只有 ES/Redis/MySQL/MongoDB），本页 Rabbit binder 一切运行行为均未运行，只做配置与源码推演。
Stream 应用模型与 binder SPI
Stream 把“应用”抽象为输入与输出绑定：Binder<T, C extends ConsumerProperties, P extends ProducerProperties> 只暴露 bindConsumer(name, group, target, properties) 与 bindProducer(name, target, properties) 两组方法，broker 差异全部下沉到实现。业务侧看到的是 MessageChannel；binder 侧负责创建 destination、消费者端点、错误基础设施与生命周期。
functional model 下绑定名由约定推导：BindableFunctionProxyFactory 的注释写明 <function-definition> + <in/out> + <index>，源码把 FunctionConstants.DELIMITER="-"、DEFAULT_INPUT_SUFFIX="in"、DEFAULT_OUTPUT_SUFFIX="out" 拼成 uppercase-in-0、uppercase-out-0（core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java:157-178；FunctionConstants.java:38,43,48）。spring.cloud.stream.bindings.<bindingName> 的属性键必须用这个推导名（或经 spring.cloud.stream.function.bindings.<derived>=<alias> 重命名），拼错不会报错、只是不生效。
BindingProperties 的四个核心字段：destination（broker 上的物理目标，缺省等于 binding 名，见 BindingServiceProperties.java:275-281）、group（消费组）、contentType（缺省 application/json，BindingProperties.java:76）、binder（绑定级 binder 名，BindingProperties.java:82）。MessageConverterConfigurer 按入站/出站方向给 channel 加内容类型拦截器：入站加 InboundContentTypeEnhancingInterceptor，出站非 functional 路径加 OutboundContentTypeConvertingInterceptor；functional 路径下函数输出由框架直接转换、不再叠加出站拦截器。contentType 配错不会让消息消失，而是让转换在错误的边界失败。
原码：consumer bind 的骨架——先向 provisioning provider 申请 destination，再创建消费者端点并挂到输入 channel：
public final Binding<MessageChannel> doBindConsumer(String name, String group,
MessageChannel inputChannel, final C properties) throws BinderException {
MessageProducer consumerEndpoint = null;
try {
ConsumerDestination destination = this.provisioningProvider
.provisionConsumerDestination(name, group, properties);
if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
enhanceMessageChannel(inputChannel);
}
consumerEndpoint = createConsumerEndpoint(destination, group, properties);
consumerEndpoint.setOutputChannel(inputChannel);
this.consumerCustomizer.configure(consumerEndpoint, name, group);
if (consumerEndpoint instanceof InitializingBean initializingConsumerEndpoint) {
initializingConsumerEndpoint.afterPropertiesSet();
}
if (properties.isAutoStartup() && consumerEndpoint instanceof Lifecycle consumerEndpointWithLifecycle) {
consumerEndpointWithLifecycle.start();
}DefaultBinderFactory 的选择逻辑决定“这条 binding 用哪个 binder”：无 binder 参数时依次看 spring.cloud.stream.defaultBinder、唯一 default candidate、bindingTargetType 可赋值的候选；多于一个候选且无默认时抛 IllegalStateException（DefaultBinderFactory.java:272-330）。选择发生在 binding 建立时，配置错误表现为启动失败或绑定到错误的 broker，而不是运行时降级。
functional model、functionDefinition 与 StreamBridge
5.x 的声明方式只剩 functional model：注册 Function/Consumer/Supplier bean，用 spring.cloud.function.definition 指定启用哪些函数（FunctionProperties.java:46,61：前缀 spring.cloud.function、键 .definition）。@EnableBinding/@StreamListener/StreamProcessor 在 Stream 5.0.4 源码树中已不存在；@StreamRetryTemplate 等其他注解仍存在。FunctionConfiguration.java 的注释提到“EnableBinding 的等价工作”（FunctionConfiguration.java:885）。旧教程里的注解式绑定不能照搬。
StreamBridge 用于“不在启动时声明的动态输出”：第一次 send(bindingName, ...) 时为该 binding 创建输出 channel 与 producer binding，缓存进 LRU（默认 dynamicDestinationCacheSize=10，BindingServiceProperties.java:131；淘汰时解绑 producer，StreamBridge.java:150-165）。send 返回 boolean——这是 MessageChannel.send 的返回值，表示本进程内 channel 接受了消息；默认 Kafka producer sync=false 时不等待发送 Future 的 broker 确认；默认同步 channel 分派下，sync=true 会等待 Future 按所配 acks 策略完成；若 StreamBridge.setAsync(true)，send先返回执行器接纳结果，而该handler稍后才等待，仍不代表业务完成（映射关系见后文“框架词义 vs broker 实际保证”）。向 input binding 发送数据会触发源码告警：You seem to be sending data to the input binding... bypassing the binder（StreamBridge.java:276-279，GH-2563）——这绕过了 binder 的错误基础设施与确认路径。
原码：StreamBridge.send 的完整路径——解析 destination、按 contentType 取函数、加锁调用、最后 messageChannel.send：
@Override
@SuppressWarnings({ "unchecked"})
public boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType) {
if (!this.initialized) {
this.afterSingletonsInstantiated();
}
ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName);
MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName);
Function functionToInvoke = this.getStreamBridgeFunction(bindingName, outputContentType.toString(), producerProperties);
if (producerProperties != null && producerProperties.isPartitioned()) {
functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);
}
// String targetType = this.resolveBinderTargetType(bindingName, binderName, MessageChannel.class,
// this.applicationContext.getBean(BinderFactory.class));
Message<?> messageToSend = data instanceof Message messageData
? MessageBuilder.fromMessage(messageData).build()
: new GenericMessage<>(data);
Message<?> resultMessage;
lock.lock();
try {
resultMessage = (Message<byte[]>) functionToInvoke.apply(messageToSend);
}
finally {
lock.unlock();
}
if (resultMessage == null) {
if (((Message) messageToSend).getPayload().getClass().getName().equals("org.springframework.kafka.support.KafkaNull")) {
resultMessage = messageToSend;
}
else {
throw new RuntimeException(functionToInvoke.getClass().getName() + " returned null");
}
}
resultMessage = (Message<?>) this.functionInvocationHelper.postProcessResult(resultMessage, null);
return messageChannel.send(resultMessage);
}动态 destination 的解析在 resolveDestination 中：已存在同名 bean 时复用（含 input binding 告警分支），否则新建 DirectWithAttributesChannel 并调用 bindingService.bindProducer(...) 后放入缓存（StreamBridge.java:265-310）。多 binder 场景下 send 的 binderName 参数指定用哪个 binder；但若 destination 已由 spring.cloud.stream.output-bindings 或先前的 binding 建立，binderName 参数不生效（官方文档 producing-and-consuming-messages.adoc 的 StreamBridge 一节）。
从依赖到 binding 的完整接线
下面是教学应用，演示接线与错误输入，不是本次运行日志。可执行验证入口在 cloud-functions 实验，由测试创建独有的 embedded Kafka 并把地址注入应用，不要求启动或改动用户的 broker。若独立复制本节为 Maven 项目，文件布局为 pom.xml、src/main/java/study/functions/Functions.java、src/main/resources/application.yml；其中 Kafka 地址由启动方提供，未给出地址就应启动失败，不能悄悄连接默认 localhost。
pom.xml 固定教学组合；Function context 是 Stream 的依赖，此处显式列出使依赖意图可见：
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>4.0.9</version>
<relativePath/>
</parent>
<groupId>study</groupId><artifactId>functions-demo</artifactId><version>1.0.0</version>
<properties><java.version>21</java.version></properties>
<dependencyManagement><dependencies><dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>2025.1.4</version><type>pom</type><scope>import</scope>
</dependency></dependencies></dependencyManagement>
<dependencies>
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-stream-binder-kafka</artifactId></dependency>
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-function-context</artifactId></dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency>
</dependencies>
<build><plugins><plugin>
<groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId>
</plugin></plugins></build>
</project>// src/main/java/study/functions/Functions.java；教学代码。
package study.functions;
import java.util.Locale;
import java.util.function.Function;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class Functions {
public static void main(String[] args) {
SpringApplication.run(Functions.class, args);
}
@Bean
public Function<String, String> uppercase() {
return value -> {
if (value.isBlank()) throw new IllegalArgumentException("blank input");
return value.toUpperCase(Locale.ROOT);
};
}
@Bean
public Function<String, String> exclaim() {
return value -> value + "!";
}
}# src/main/resources/application.yml；仅 uppercase 被绑定，exclaim 只进入 catalog。
spring:
application:
name: u36-functions-demo
main:
web-application-type: none
cloud:
function:
definition: uppercase
stream:
bindings:
uppercase-in-0:
destination: ${LAB_INPUT_TOPIC}
group: ${LAB_CONSUMER_GROUP}
content-type: text/plain
consumer:
max-attempts: 3
back-off-initial-interval: 1000
back-off-multiplier: 2.0
back-off-max-interval: 10000
uppercase-out-0:
destination: ${LAB_OUTPUT_TOPIC}
content-type: text/plain
kafka:
binder:
brokers: ${LAB_KAFKA_BOOTSTRAP_SERVERS}
bindings:
uppercase-in-0:
consumer:
ack-mode: RECORD
start-offset: earliest
enable-dlq: true
dlq-name: ${LAB_DLQ_TOPIC}
uppercase-out-0:
producer:
sync: false这些属性在创建 binding/消费者容器时读取，本例不承诺普通 Environment 刷新会重建 binding。destination 与 group 是身份，不是展示标题；${LAB_*} 由本次实验启动方生成并注入，四个 topic/组名使用本次独有前缀。content-type=text/plain 匹配字符串；若改为 JSON POJO，就要同时改函数签名、载荷与转换配置。sync=false 是 Kafka producer 扩展属性，此时 send 返回与异步 broker 确认必须分别观察；改为 sync=true 后，还需检查该 binder 的同步等待与超时路径，不能沿用异步发送的时点结论。
用字符串 hello 作为输入，预期输出 HELLO；用三个空格作为输入，uppercase 连续抛错，预期进入配置的 DLQ。这里的“预期”仅说明教学例子：观察时要同时查输出/DLQ 记录、函数失败、源消费组 committed offset，不能只查一条应用日志。把键误写成 uppercaseInput 的反例不会修改 uppercase-in-0；诊断从启动时的函数 definition、实际 binding 名、topic/组名和有效属性开始，修复键名后用一个新的合成事件 id重试，避免把旧位点误判成修复无效。独立项目可用 mvn spring-boot:run 启动，但本仓库的自动验证统一使用 README 的公开 runner，由 runner 管理临时 broker 和清理。
contentType 的转换边界在 MessageConverterConfigurer.configureMessageChannel：它根据 binding 的 contentType、方向与是否 functional 决定加哪类拦截器：
private void configureMessageChannel(MessageChannel channel, String channelName,
boolean inbound) {
Assert.isAssignable(AbstractMessageChannel.class, channel.getClass());
AbstractMessageChannel messageChannel = (AbstractMessageChannel) channel;
BindingProperties bindingProperties = this.bindingServiceProperties
.getBindingProperties(channelName);
String contentType = bindingProperties.getContentType();
ProducerProperties producerProperties = bindingProperties.getProducer();
boolean partitioned = !inbound && producerProperties != null && producerProperties.isPartitioned();
boolean functional = streamFunctionProperties != null
&& (StringUtils.hasText(streamFunctionProperties.getDefinition())
|| StringUtils.hasText(bindingServiceProperties.getInputBindings())
|| StringUtils.hasText(bindingServiceProperties.getOutputBindings()));
if (partitioned) {
if (inbound || !functional) {
messageChannel.addInterceptor(new PartitioningInterceptor(bindingProperties));
}
}
Environment environment = this.beanFactory == null ? null : this.beanFactory.getBean(Environment.class);
ConsumerProperties consumerProperties = bindingProperties.getConsumer();
if (this.isNativeEncodingNotSet(producerProperties, consumerProperties, inbound)) {
if (inbound) {
messageChannel.addInterceptor(new InboundContentTypeEnhancingInterceptor(contentType));
}
else {
if (environment != null && environment.containsProperty("spring.cloud.stream.rabbit.bindings." + channelName + ".producer.routing-key-expression")) {
Map<String, String> channelNameToFunctions = BindableFunctionProxyFactory.getChannelNameToFunctions();
String functionName = channelNameToFunctions.get(channelName);
String outputBindings = environment.getProperty("spring.cloud.stream.output-bindings");
FunctionInvocationWrapper function = null;
if (outputBindings != null && outputBindings.contains(functionName)) {
try {
function = retrieveFunction(functionName);
}
catch (Exception e) {
// we expect an exception in this case since we didn't have a matching function
// for the function name due to output-bindings, hence passing through.
// See https://github.com/spring-cloud/spring-cloud-stream/issues/2921 for more details.
}
}
else {
function = retrieveFunction(functionName);
}
if (function != null) {
function.setSkipOutputConversion(true);
}
functional = false;
}
if (!functional) {
messageChannel.addInterceptor(new OutboundContentTypeConvertingInterceptor(contentType, this.compositeMessageConverter));
}
}
}
}
private FunctionInvocationWrapper retrieveFunction(String functionName) {
FunctionCatalog catalog = this.beanFactory.getBean(FunctionCatalog.class);
FunctionInvocationWrapper function = catalog.lookup(functionName);
if (function != null) {
function.setSkipOutputConversion(true);
}
return function;
}Kafka同步发送分支明确调用setSync，正是返回值必须带配置条件的原因：
}
setMessageKeyExpression(messageKeyExpression);
setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory());
if (producerProperties.isPartitioned()) {
setPartitionIdExpression(PARSER.parseExpression(
"headers['" + BinderHeaders.PARTITION_HEADER + "']"));
}
if (producerProperties.getExtension().isSync()) {
setSync(true);
}
if (producerProperties.getExtension().getSendTimeoutExpression() != null) {
setSendTimeoutExpression(producerProperties.getExtension().getSendTimeoutExpression());
}
this.producerFactory = producerFactory;起点、组与广播/竞争
这是 binder 语义权威（自 U34 迁移而来；U34 页保留 Bus 叙事并回指本页）。Kafka binder 对新消费组的起点：显式配置了 spring.cloud.stream.bindings.<name>.group 时 startOffset 默认 earliest；匿名组默认 latest（上游文档 docs/modules/ROOT/pages/kafka/kafka_overview.adoc:241-244：“If the consumer group is set explicitly... 'startOffset' is set to earliest. Otherwise, it is set to latest for the anonymous consumer group.”）。startOffset 只影响没有已提交位点的新组；已有位点的组从位点继续，改它不会重放历史。resetOffsets=true 才会对所有新分配分区执行 seekToBeginning/seekToEnd（同文档 :351-369），且与 KafkaBindingRebalanceListener 互斥。
消费组是持久身份，决定“广播还是竞争”：两个实例共用同一显式 group → 分区在成员间分摊，每条消息只被其中一个实例消费（竞争）；每个实例用独立 group → 各自收到全量（广播）。匿名组随实例生灭，重启即新组，按 latest 只看到启动后的新消息。这正是 U34 Bus 定向刷新“共用显式组竞争分区、广播需每实例独立组”一句话的机制出处；Config与Bus：配置读取、传播与版本回滚 的 Bus 寻址、唯一 id 与真实 Kafka 传播 一节保留 Bus 叙事，本页接管 binder 语义。
resetOffsets 与 startOffset 的完整字段集中在 KafkaConsumerProperties：
/**
* Controlling the container acknowledgement mode. This is the preferred way to control the ack mode on the
* container instead of the deprecated autoCommitOffset property.
*/
private ContainerProperties.AckMode ackMode;
/**
* Consumer ack count. When set, the container will send an acknowledgment after the specified number
* of messages have been consumed. Only applicable for COUNT or COUNT_TIME ack modes.
* <p>
* This property is binding-level and takes precedence over the global
* {@code spring.kafka.listener.ackCount} when explicitly configured on the binding.
*/
private Integer ackCount;
/**
* Consumer ack time. When set, the container will send an acknowledgment after the specified duration
* has elapsed since the last message was consumed. Only applicable for TIME or COUNT_TIME ack modes.
* <p>
* This property is binding-level and takes precedence over the global
* {@code spring.kafka.listener.ackTime} when explicitly configured on the binding.
*/
private Duration ackTime;
/**
* Flag to enable auto commit on error in polled consumers.
*/
private Boolean autoCommitOnError;
/**
* The starting offset for new groups. Allowed values: earliest and latest.
*/
private StartOffset startOffset;
/**
* Whether to reset offsets on the consumer to the value provided by startOffset.
* Must be false if a KafkaRebalanceListener is provided.
*/
private boolean resetOffsets;
/**
* When set to true, it enables DLQ behavior for the consumer.
* By default, messages that result in errors are forwarded to a topic named error.name-of-destination.name-of-group.
* The DLQ topic name can be configurable by setting the dlqName property.
*/
private boolean enableDlq;ackMode 与提交前沿
Spring Cloud Stream 的 ackMode 就是 Spring Kafka 容器的 ContainerProperties.AckMode，Kafka binder 在建立容器时把它设进去，设置条件为 !batchMode || ackMode != RECORD，即只排除 batch listener 与 RECORD 的组合：
ContainerProperties.AckMode ackMode = extendedConsumerProperties.getExtension().getAckMode();
if (ackMode != null) {
if (!extendedConsumerProperties.isBatchMode() || ackMode != ContainerProperties.AckMode.RECORD) {
messageListenerContainer.getContainerProperties()
.setAckMode(ackMode);
}
}
// Apply binding-level ackCount and ackTime after global propagation
// These take precedence over global spring.kafka.listener values when explicitly set
KafkaConsumerProperties consumerExt = extendedConsumerProperties.getExtension();
Integer bindingAckCount = consumerExt.getAckCount();
if (bindingAckCount != null) {
messageListenerContainer.getContainerProperties().setAckCount(bindingAckCount);
}
Duration bindingAckTime = consumerExt.getAckTime();
if (bindingAckTime != null) {
messageListenerContainer.getContainerProperties().setAckTime(bindingAckTime.toMillis());
}ackMode 字段本身的定义与注释（“控制容器确认模式，取代已弃用的 autoCommitOffset”）在 KafkaConsumerProperties.java:97-101；autoCommitOffset 自 3.1 起弃用，bindings 级 ackCount/ackTime 只在 COUNT/COUNT_TIME 模式生效（同文件 :103-124）。常用值：BATCH（poll 批次处理完再提交）、RECORD（每条处理完提交）、MANUAL/MANUAL_IMMEDIATE（应用通过 Acknowledgment 显式确认）。
把这三个值映射到 Kafka消费位点与重平衡 的三种进度（poll position / 业务完成 / committed offset）：ackMode 只决定“容器在什么时机把已交付前缀提交为 committed offset”，不改变三种进度之分，也不把“提交”变成“业务成功”。
| ackMode | 提交时机 | 失败时的安全前沿 |
|---|---|---|
| BATCH | poll 返回记录的处理完成后按容器策略提交 | 还取决于 record/batch listener、error handler 的 recover/seek 与事务；不承诺失败必然整批重投 |
| RECORD | 每条处理完成或被配置的恢复器处置后提交 | 失败未被恢复时可停在失败前；恢复器吞下错误后也可推进，见无 DLQ 实测负控 |
| MANUAL | 应用调用 acknowledge() 后按容器时机提交 | 忘记确认使提交滞后，重启/重平衡后可能重复；不等于当前进程立即自动重投 |
BATCH 不是错误处置策略。record listener 配 DefaultErrorHandler 时，容器可能先提交失败前已处理的前缀，再重投剩余；batch listener、事务回滚与自定义 recoverer 又不同。MANUAL 下提前确认则可能放弃尚未完成的业务工作；不确认通常只使位点不推进。可靠并行消费必须维护每分区有序交付队列，只推进连续完成前缀，算法权威见 U57 的“三种进度和一个空洞”。本票只实测非事务 RECORD 路径，不把它外推到其他模式。
显式确认的用法与观测
若业务需要显式确认，可把教学应用的函数定义改为 receive，删除原函数输出 binding，新增以下 bean，并把 spring.cloud.stream.kafka.bindings.receive-in-0.consumer.ack-mode 设为 MANUAL。receive-in-0 的 destination/group/content-type 仍按上一节的完整配置设置。此例只演示同步处理与确认先后，不实现生产幂等：
// 放入 Functions 类；所用 import：java.util.function.Consumer、
// org.springframework.messaging.Message、org.springframework.kafka.support.KafkaHeaders、
// org.springframework.kafka.support.Acknowledgment。
@Bean
public Consumer<Message<String>> receive() {
return message -> {
Acknowledgment ack = message.getHeaders()
.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
if (ack == null) throw new IllegalStateException("MANUAL ack header missing");
String value = message.getPayload();
if (value.isBlank()) throw new IllegalArgumentException("blank input");
System.out.println("synthetic work finished: " + value);
ack.acknowledge(); // 同步工作结束后才确认；真实库提交与此调用不在同一事务。
};
}MANUAL 的 acknowledge() 是向容器声明可提交，提交可仍由 poll 批次规则推进；MANUAL_IMMEDIATE 的立即提交还取决于是否在消费者线程调用。把工作交给线程池后立刻确认的反例是：进程在池内任务完成前退出，恢复从已提交位点开始，任务没有被重放。修复时让同步处理返回前完成业务提交，或另行设计按分区连续完成前缀与有界在途队列，并独立验证退出时序；不能只把 MANUAL 改名为 MANUAL_IMMEDIATE。
故障诊断至少保留四项：消息的 topic/partition/offset、处理结果、确认模式及错误处理器、消费组 committed offset。poll position 超前只能说明客户端已拉取；“未调用 acknowledge”本身也不能推出无限重投，因为错误处理器的恢复/跳过策略同样参与位点推进。修复后必须重启同组消费者，核对失败记录是否被重投及业务效果是否重复；三个进度的完整解释仍以 U57 的权威正文为准。
重试、DLQ 与客户端边界
消费者处理函数抛异常后，重试发生在两个可能的层级：binder 内（框架 RetryTemplate，默认 maxAttempts=3、初始退避 1000ms、乘数 2.0、最大间隔 10000ms）或监听器容器内（Spring Kafka DefaultAfterRollbackProcessor / DefaultErrorHandler）。ConsumerProperties 的注释明确这些是“framework provided RetryTemplate”，也可注册带 @StreamRetryTemplate 限定的自定义 RetryTemplate，并由 binding 的 consumer.retry-template-name 选择：
/**
* The number of attempts to process the message (including the first) in the event of
* processing failures. This is a RetryTemplate configuration which is provided by the
* framework. Default: 3. Set to 1 to disable retry. You can also provide custom
* RetryTemplate in the event you want to take complete control of the RetryTemplate.
* Simply configure it as @Bean inside your application configuration.
*/
private int maxAttempts = 3;
/**
* The backoff initial interval on retry. This is a RetryTemplate configuration which
* is provided by the framework. Default: 1000 ms. You can also provide custom
* RetryTemplate in the event you want to take complete control of the RetryTemplate.
* Simply configure it as @Bean inside your application configuration.
*/
private int backOffInitialInterval = 1000;
/**
* The maximum backoff interval. This is a RetryTemplate configuration which is
* provided by the framework. Default: 10000 ms. You can also provide custom
* RetryTemplate in the event you want to take complete control of the RetryTemplate.
* Simply configure it as @Bean inside your application configuration.
*/
private int backOffMaxInterval = 10000;
/**
* The backoff multiplier.This is a RetryTemplate configuration which is provided by
* the framework. Default: 2.0. You can also provide custom RetryTemplate in the event
* you want to take complete control of the RetryTemplate. Simply configure it
* as @Bean inside your application configuration.
*/
private double backOffMultiplier = 2.0;
/**
* Whether exceptions thrown by the listener that are not listed in the
* 'retryableExceptions' are retryable.
*/
private boolean defaultRetryable = true;
/**
* Allows you to further qualify which RetryTemplate to use for a specific consumer
* binding..
*/
private String retryTemplateName;
/**
* A map of Throwable class names in the key and a boolean in the value. Specify those
* exceptions (and subclasses) that will or won't be retried.
*/
private Map<Class<? extends Throwable>, Boolean> retryableExceptions = new LinkedHashMap<>();AbstractBinder.buildRetryTemplate 把 maxAttempts 翻译成 maxRetries = maxAttempts - 1，把 backOff* 翻译成延迟、乘数、最大间隔，并把 retryableExceptions 映射为 include/exclude；普通 RetryTemplate bean 不会自动接管；源码注入带 @StreamRetryTemplate 限定的 Map，指定 retryTemplateName 时按名选择，未指定且 Map 非空时取首个：
/**
* Create and configure a default retry template unless one has already been provided
* via @Bean by an application.
* @param properties The properties.
* @return The retry template
*/
protected RetryTemplate buildRetryTemplate(ConsumerProperties properties) {
RetryTemplate rt;
if (CollectionUtils.isEmpty(this.consumerBindingRetryTemplates)) {
rt = new RetryTemplate();
Map<Class<? extends Throwable>, Boolean> retryableExceptionMapping = properties.getRetryableExceptions();
List<Class<? extends Throwable>> retryableExceptions = new ArrayList<>();
List<Class<? extends Throwable>> nonRetryableExceptions = new ArrayList<>();
for (var classBooleanEntry : retryableExceptionMapping.entrySet()) {
Class<? extends Throwable> exceptionClass = classBooleanEntry.getKey();
if (classBooleanEntry.getValue()) {
retryableExceptions.add(exceptionClass);
}
else {
nonRetryableExceptions.add(exceptionClass);
}
}
RetryPolicy retryPolicy =
RetryPolicy.builder()
.maxRetries(Math.max(0, properties.getMaxAttempts() - 1))
.delay(Duration.ofMillis(properties.getBackOffInitialInterval()))
.multiplier(properties.getBackOffMultiplier())
.maxDelay(Duration.ofMillis(properties.getBackOffMaxInterval()))
.includes(retryableExceptions)
.excludes(nonRetryableExceptions)
.build();
rt.setRetryPolicy(retryPolicy);
}
else {
rt = StringUtils.hasText(properties.getRetryTemplateName())
? this.consumerBindingRetryTemplates
.get(properties.getRetryTemplateName())
: this.consumerBindingRetryTemplates.values().iterator().next();
}
return rt;
}Kafka binder 的装配条件（KafkaMessageChannelBinder.java:742-753）：非 batch、maxAttempts>1 且无事务管理器时，给适配器设 RetryTemplate 与 recovery callback（指向错误基础设施的 recoverer）；若 enableDlq=false，容器挂一个 DefaultErrorHandler(new FixedBackOff(0L, 0L))（不重试）。batch 模式下 binder 内重试被强制关闭（上游文档 kafka_overview.adoc:384：“Retry within the binder is not supported when using batch mode, so maxAttempts will be overridden to 1.”），批量重试要靠 ListenerContainerCustomizer 配 DefaultErrorHandler 或 Acknowledgment.nack(index, sleep) 的部分批提交。
enableDlq=true 时，失败记录被 republish 到 DLQ topic，默认名 error.<destination>.<group>（源码 KafkaMessageChannelBinder.java:1396-1409 与文档 kafka_overview.adoc:248-279 一致），可用 dlqName 或 DlqDestinationResolver 覆盖。DLQ 头带 x-original-topic、x-exception-message、x-exception-stacktrace（byte[]），默认投到与原记录相同的分区号。getErrorMessageHandler 只在非“batch + 自定义容器”组合下返回 DLQ handler：
@SuppressWarnings("unchecked")
@Override
protected MessageHandler getErrorMessageHandler(final ConsumerDestination destination,
final String group,
final ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
KafkaConsumerProperties kafkaConsumerProperties = properties.getExtension();
if (kafkaConsumerProperties.isEnableDlq() && !isBatchAndListenerContainerWithDlqAndRetryCustomizer(properties)) {
KafkaProducerProperties dlqProducerProperties = kafkaConsumerProperties
.getDlqProducerProperties();
KafkaAwareTransactionManager<byte[], byte[]> transMan = transactionManager(
properties.getExtension().getTransactionManager());
final ExtendedProducerProperties<KafkaProducerProperties> producerProperties =
new ExtendedProducerProperties<>(dlqProducerProperties);
producerProperties.populateBindingName(properties.getBindingName());
ProducerFactory<?, ?> producerFactory = transMan != null
? transMan.getProducerFactory()
: getProducerFactory(null, producerProperties,
destination.getName() + ".dlq.producer", destination.getName());
final KafkaTemplate<?, ?> kafkaTemplate = new KafkaTemplate<>(producerFactory);
kafkaTemplate.setObservationEnabled(this.configurationProperties.isEnableObservation());
kafkaTemplate.setKafkaAdmin(this.kafkaAdmin);
Object timeout = producerFactory.getConfigurationProperties().get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG);
Long sendTimeout = null;
if (timeout instanceof Number timeoutAsNumber) {
sendTimeout = timeoutAsNumber.longValue() + 2000L;
}
else if (timeout instanceof String timeoutAsString) {
sendTimeout = Long.parseLong(timeoutAsString) + 2000L;
}
if (timeout == null) {
sendTimeout = ((Integer) ProducerConfig.configDef()
.defaultValues()
.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG)).longValue() + 2000L;
}
@SuppressWarnings("rawtypes")
DlqSender<?, ?> dlqSender = new DlqSender(kafkaTemplate, sendTimeout);
return (message) -> {
List<ConsumerRecord<Object, Object>> records;
if (!properties.isBatchMode()) {
ConsumerRecord<Object, Object> record = StaticMessageHeaderAccessor.getSourceData(message);
records = List.of(Objects.requireNonNull(record));
}
else {
records = StaticMessageHeaderAccessor.getSourceData(message);
}
if (!CollectionUtils.isEmpty(records)) {
records.forEach(record ->
handleRecordForDlq(record, destination, group, properties, kafkaConsumerProperties,
dlqProducerProperties, transMan, dlqSender, message));
}
};
}
return null;
}这不是 broker 协议。 重试、死信、延迟消息与消息积压 的 三条重试路径及其失败窗口 一节原文：“Kafka日志没有一个通用的‘把此条nack后延迟10秒再投递’协议API”。Stream binder 的 maxAttempts/enableDlq 是客户端行为：框架在消费循环里重试、把失败记录用 KafkaTemplate 重新生产到 DLQ topic、然后推进源位点。由此有三个工程结论：重试总时长超过 max.poll.interval.ms 会触发 rebalance（文档建议此时用 ListenerContainerWithDlqAndRetryCustomizer.retryAndDlqInBinding=false 把重试下放到容器，并配 DeadLetterPublishingRecoverer）；在非事务路径，DLQ 发布与源位点推进不属于一个原子提交；Kafka 事务路径可以把 Kafka 内的输出和 offset 纳入同一事务，但不能扩展到外部数据库；DLQ 是“业务状态”，不是“数据不丢”的证明，回放与幂等归 U61。
协议式的重试路径（retry topic、调度表）仍是 U57 的权威；本页只声明 binder 默认走的是“客户端重投 + republish”路径。
重试参数的范围、时点与失败定位
以下前缀为 spring.cloud.stream.bindings.<binding>.consumer.，均在 binding 创建时配置。它们是框架消费重试参数，与 producer 重试、Kafka 协议重传或业务定时回放分开：
| 参数 | 本版默认/本例值 | 含义与限制 |
|---|---|---|
max-attempts | 3 / 3 | 包含首次，所以最多再试两次；1 关闭 binder 内重试，不自动关闭容器错误处理器的重试 |
back-off-initial-interval | 1000 / 1000 | 首次等待的毫秒数，耗在消费处理路径上 |
back-off-multiplier | 2.0 / 2.0 | 后续等待的倍率；三次总尝试的等待为 1s、2s，累计等待 3s |
back-off-max-interval | 10000 / 10000 | 单次等待上限，单位毫秒，不是整条消息的总截止时间 |
retry-template-name | 无 / 未设 | 选择带 @StreamRetryTemplate 的具名模板；普通 RetryTemplate bean 不等于被 binder 选中 |
Kafka 扩展项要写在 spring.cloud.stream.kafka.bindings.<binding>.consumer.：enable-dlq 默认 false，本例显式 true；dlq-name 本例显式指定独有 topic。把它们写到通用 bindings.<binding>.consumer 前缀，是“yaml 看似配了但 DLQ 不出现”的另一反例。排障先确认异常是否从命令式函数逸出、错误处理器是否被自定义替换、group 是否显式、DLQ topic/分区与授权是否正确，再看 republish 发送结果和源位点。若 payload 的反序列化发生在函数之前，函数体内的 try/catch 也接不住；应从 binder/container 的反序列化与错误通道配置定位。
修复不能止于“DLQ 里有一条”：取出原记录身份、异常和失败次数，修复数据/代码后向独有回放目标投递，核对业务去重键及源/目标记录；回放操作、允许重试次数与最终放弃/人工处理由应用负责人承担。回放是新一次生产，可能再次重复，业务状态机及 Outbox/Inbox 的实现由 U61 负责。
自定义模板的注入限定也属于机制，而非任意同类型Bean：
private volatile EvaluationContext evaluationContext;
@Autowired(required = false)
@StreamRetryTemplate
private Map<String, RetryTemplate> consumerBindingRetryTemplates;发送侧与消费侧错误出口
消费侧先区分 binding 错误 channel、全局 errorChannel 和 DLQ topic。binding channel 的名字由 <binderIdentity>.<bindingName>.errors 生成（AbstractMessageChannelBinder.java:1090-1105）；error.<destination>.<group> 是默认 DLQ topic 名，二者不可混用。registerErrorInfrastructure 解析 error-handler-definition，创建/复用 binding channel 与 recoverer，再选择应用处理器或 binder 处理器，并按条件桥接全局 channel；不是三层按顺序重试，更不是每层各有一次确认：
protected final ErrorInfrastructure registerErrorInfrastructure(
ConsumerDestination destination, String group, C consumerProperties,
boolean polled) {
ErrorMessageStrategy errorMessageStrategy = getErrorMessageStrategy();
String errorChannelName = errorsBaseName(destination, group, consumerProperties);
BindingServiceProperties bsp = this.getBindingServiceProperties();
FunctionInvocationWrapper userErrorHandler = null;
String errorHandlerDefinition = null;
if (bsp != null && StringUtils.hasText(consumerProperties.getBindingName())) {
BindingProperties bp = bsp.getBindingProperties(consumerProperties.getBindingName());
errorHandlerDefinition = bp.getErrorHandlerDefinition();
FunctionCatalog catalog = getApplicationContext().getBean(FunctionCatalog.class);
if (StringUtils.hasText(errorHandlerDefinition)) {
userErrorHandler = catalog.lookup(errorHandlerDefinition);
if (!(userErrorHandler != null && userErrorHandler.getFunctionDefinition().equals(errorHandlerDefinition))) {
userErrorHandler = null;
}
}
}
AbstractSubscribableChannel binderErrorChannel;
if (this.getApplicationContext().containsBean(errorChannelName)) {
binderErrorChannel = this.getApplicationContext().getBean(errorChannelName, AbstractSubscribableChannel.class);
}
else {
binderErrorChannel = new BinderErrorChannel();
binderErrorChannel.setComponentName(errorChannelName);
((GenericApplicationContext) getApplicationContext()).registerBean(
errorChannelName, SubscribableChannel.class, () -> binderErrorChannel);
}
boolean userHandlerSubscribed = this.subscribeFunctionErrorHandler(errorChannelName, consumerProperties.getBindingName());
ErrorMessageSendingRecoverer recoverer = new ErrorMessageSendingRecoverer(binderErrorChannel, errorMessageStrategy);
String recovererBeanName = getErrorRecovererName(destination, group, consumerProperties);
if (!getApplicationContext().containsBean(recovererBeanName)) {
((GenericApplicationContext) getApplicationContext()).registerBean(
recovererBeanName, ErrorMessageSendingRecoverer.class, () -> recoverer);
}
MessageHandler binderProvidedErrorHandler = polled
? getPolledConsumerErrorMessageHandler(destination, group, consumerProperties)
: getErrorMessageHandler(destination, group, consumerProperties);
String errorMessageHandlerName = getErrorMessageHandlerName(destination, group,
consumerProperties);
if (binderProvidedErrorHandler == null) {
binderProvidedErrorHandler = this.getDefaultErrorMessageHandler(binderErrorChannel, polled);
}
if (binderProvidedErrorHandler != null && !userHandlerSubscribed) {
if (this.isSubscribable(binderErrorChannel) && !getApplicationContext().containsBean(errorMessageHandlerName)) {
MessageHandler h = binderProvidedErrorHandler;
((GenericApplicationContext) getApplicationContext()).registerBean(errorMessageHandlerName, MessageHandler.class, () -> h);
binderErrorChannel.subscribe(binderProvidedErrorHandler);
}
else {
this.logger.warn("The provided errorChannel '" + errorChannelName
+ "' is an instance of DirectChannel, "
+ "so no more subscribers could be added which may affect DLQ processing. "
+ "Resolution: Configure your own errorChannel as "
+ "an instance of PublishSubscribeChannel");
}
}
// Setup a bridge to global errorChannel to ensure logging of errors could be controlled via standard SI way
if (this.getApplicationContext().containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) && this.isSubscribable(binderErrorChannel)) {
String errorBridgeHandlerName = getErrorBridgeName(destination, group, consumerProperties);
if (!getApplicationContext().containsBean(errorBridgeHandlerName)) {
SubscribableChannel globalErrorChannel = this.getApplicationContext().getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class);
BridgeHandler bridge = new BridgeHandler();
bridge.setOutputChannel(globalErrorChannel);
binderErrorChannel.subscribe(bridge);
((GenericApplicationContext) getApplicationContext()).registerBean(errorBridgeHandlerName, BridgeHandler.class, () -> bridge);
}
}
return new ErrorInfrastructure(binderErrorChannel, recoverer, binderProvidedErrorHandler);
}ErrorMessage 携带原始消息（getOriginalMessage()）与异常，因此自定义 handler 能看到触发错误的那条记录——这也是为什么“订阅 errorChannel 后消息仍会重投/进 DLQ”是常见误解：仅仅存在 error channel 订阅，不能证明确认、重投或DLQ已经发生；仍需看订阅者是否传播异常、恢复器和容器策略；只有应用侧显式配置的 error-handler-definition（Consumer<ErrorMessage>）会替换掉框架提供的错误处理器，且与 DLQ 互斥（官方 overview-error-handling.adoc：“When you provide a custom error handler, sending failed messages to DLQ won't work even if it is configured to do so.”）。
发送侧的错误出口不同：StreamBridge.send 可返回 false 或抛异常；含义取决于 channel 与 producer 配置。默认 Kafka producer.sync=false 下，true 只表示本次 channel 调用接受，broker 异步失败还需发送结果/错误观测；默认同步channel分派时，producer.sync=true 会使调用等待 Kafka handler 的发送 Future，承诺由 acks 和超时策略限定；若StreamBridge异步分派，等待发生在执行器线程，send本身仍可能先返回。StreamBridge.setAsync(true) 另把 channel 分派到执行器，与 Kafka sync 是两个独立开关。任何组合都不能把发送成功直接当成业务提交，参见 七个成功点。
多 binder 与默认 binder
一个应用可以同时引入 Kafka 与 Rabbit binder（不同 starter），此时必须能区分每条 binding 用哪个 binder：全局 spring.cloud.stream.defaultBinder=<name>，或绑定级 spring.cloud.stream.bindings.<name>.binder=<name>（官方 multiple-binders.adoc）。DefaultBinderFactory 的无名请求在“多候选且无默认”时抛异常，异常消息直接列出可用 binder 名（DefaultBinderFactory.java:200-208）。defaultBinder 属性定义在 BindingServiceProperties.java:124，运行时经 binderConfigurations 解析。
以下为双 binder 配置推演，未运行 RabbitMQ。在前文 POM 再加入 BOM 管理的 spring-cloud-stream-binder-rabbit 后，给两个 binder 分别设置环境；kafkaSide/rabbitSide 是本应用自定义的配置名，type 才是 binder 类型，不能混用：
spring:
cloud:
function:
definition: uppercase
stream:
default-binder: kafkaSide
binders:
kafkaSide:
type: kafka
environment:
spring.cloud.stream.kafka.binder.brokers: ${LAB_KAFKA_BOOTSTRAP_SERVERS}
rabbitSide:
type: rabbit
environment:
spring.rabbitmq.host: ${LAB_RABBIT_HOST}
spring.rabbitmq.port: ${LAB_RABBIT_PORT}
spring.rabbitmq.username: ${LAB_RABBIT_USER}
spring.rabbitmq.password: ${LAB_RABBIT_PASSWORD}
bindings:
uppercase-in-0:
binder: kafkaSide
destination: ${LAB_INPUT_TOPIC}
group: ${LAB_CONSUMER_GROUP}
content-type: text/plain
uppercase-out-0:
binder: rabbitSide
destination: ${LAB_OUTPUT_EXCHANGE}
content-type: text/plain同一 uppercase 函数此时从 Kafka 消费、向 Rabbit 发布；它仍没有跨 broker 事务。Rabbit 不可达而 Kafka 函数已计算成功时，失败属于发送侧，必须按实际同步/异步模式和错误出口处理，不能由“一个函数有输入输出”推出跨 broker 原子交付。只有一个 broker 故障时另一个 binding 是否继续工作，也须实跑其生命周期和线程/连接资源后才能断言；本页不把此配置推演计入双栈运行证据。
Kafka 侧本轮有实验入口（以证据为准）；Rabbit 侧未运行，只能读配置与源码。Rabbit binder 的消费属性暴露了 broker 差异：transacted、acknowledgeMode（默认 AUTO）、prefetch（默认 1）、batchSize、republishToDlq（默认 true）、requeueRejected（默认 false）：
/**
* true to use transacted channels.
*/
private boolean transacted;
/**
* container acknowledge mode.
*/
private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
/**
* maxumum concurrency of this consumer (threads).
*/
private int maxConcurrency = 1;
/**
* number of prefetched messages pre consumer thread.
*/
private int prefetch = 1;
/**
* messages per acknowledgment (and commit when transacted).
*/
private int batchSize = 1;
/**
* true for a durable subscription.
*/
private boolean durableSubscription = true;
/**
* republish failures to the DLQ with diagnostic headers.
*/
private boolean republishToDlq = true;
/**
* when republishing to the DLQ, the delivery mode to use.
*/
private MessageDeliveryMode republishDeliveyMode = MessageDeliveryMode.PERSISTENT;
/**
* true to requeue rejected messages, false to discard (or route to DLQ).
*/
private boolean requeueRejected = false;Rabbit 的确认语义是 channel 级（AcknowledgeMode.NONE/AUTO/MANUAL + channelTransacted），与 Kafka 的分区位点完全不同——这就是 binder 抽象“统一接口、不统一保证”的典型：同为 ackMode 词，Kafka 下映射到 ContainerProperties.AckMode，Rabbit 下映射到 SimpleMessageListenerContainer 的确认模式与事务 channel：
private ObservableListenerContainer createAndConfigureContainer(ConsumerDestination consumerDestination,
String group, ExtendedConsumerProperties<RabbitConsumerProperties> properties, String destination,
RabbitConsumerProperties extension) {
if (extension.getContainerType().equals(ContainerType.STREAM)) {
return StreamUtils.createContainer(consumerDestination, group, properties, destination,
getApplicationContext());
}
boolean directContainer = extension.getContainerType()
.equals(ContainerType.DIRECT);
AbstractMessageListenerContainer listenerContainer = directContainer
? new DirectMessageListenerContainer(this.connectionFactory)
: new SimpleMessageListenerContainer(this.connectionFactory);
listenerContainer.setBeanName(consumerDestination.getName() + "." + group + ".container");
listenerContainer
.setAcknowledgeMode(extension.getAcknowledgeMode());
listenerContainer.setChannelTransacted(extension.isTransacted());
listenerContainer
.setDefaultRequeueRejected(extension.isRequeueRejected());
int concurrency = properties.getConcurrency();
concurrency = concurrency > 0 ? concurrency : 1;
if (directContainer) {
setDMLCProperties(properties,
(DirectMessageListenerContainer) listenerContainer, concurrency);
}
else {
setSMLCProperties(properties,
(SimpleMessageListenerContainer) listenerContainer, concurrency);
}
listenerContainer.setPrefetchCount(extension.getPrefetch());Rabbit 的 DLQ 是 broker 侧 DLX/TTL 机制，U58 才是权威；本页只说明 binder 的 republishToDlq=true 默认值是客户端重新发布（带诊断头到 DLX 路由），与 Kafka 的 republish 类似而非 broker 自动死信。多 binder 双栈的运行行为（同应用同时连 Kafka 与 Rabbit、绑定级切换、故障隔离）本机无法验证，未运行。
Kafka Streams binder：拓扑、状态与恢复身份
Function<String,String> 面向一次消息调用，Function<KStream<K,V>,KStream<K,R>> 面向持续拓扑。后者的输入是 Kafka Streams 创建的流，函数定义处理节点，groupBy/count/join 可以创建本地状态及对应 changelog/repartition topic。KTable 表示按 key 更新的当前状态，不能把它与“每条事件都保留的 KStream”互换。这里纳入 spring-cloud-stream-binder-kafka-streams 5.0.4 的适配边界；Kafka 日志复制与通用位点原理回指 U57，数据方向的全引擎课不在本节重复。
下面是固定源码核查后的教学配置，未运行 Kafka Streams 拓扑，不是公开 Stream 实验的已执行用例。新建独立 Boot4.0.9/Cloud2025.1.4 工程，保留本页通用 parent、Cloud BOM 与 Boot Maven 插件，只选该 binder（不要另加入普通 Kafka/Rabbit binder 来制造候选歧义）：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
<version>5.0.4</version>
</dependency>src/main/java/study/count/CountApp.java 是完整函数入口：
package study.count;
import java.util.Locale;
import java.util.function.Function;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class CountApp {
public static void main(String[] args) {
SpringApplication.run(CountApp.class, args);
}
@Bean
Function<KStream<String, String>, KStream<String, Long>> count() {
return input -> input
.filter((key, value) -> value != null && !value.isBlank())
.mapValues(value -> value.toLowerCase(Locale.ROOT))
.groupBy((key, value) -> value,
Grouped.with(Serdes.String(), Serdes.String()))
.count(Materialized.as("u36-counts"))
.toStream();
}
}src/main/resources/application.properties：
spring.application.name=u36-count
spring.main.web-application-type=none
spring.cloud.function.definition=count
spring.cloud.stream.bindings.count-in-0.destination=u36-count-input
spring.cloud.stream.bindings.count-out-0.destination=u36-count-output
spring.cloud.stream.kafka.streams.binder.brokers=${KAFKA_BOOTSTRAP:127.0.0.1:9092}
spring.cloud.stream.kafka.streams.binder.functions.count.application-id=u36-count-v1
spring.cloud.stream.kafka.streams.binder.configuration.state.dir=${java.io.tmpdir}/u36-count-v1
spring.cloud.stream.kafka.streams.bindings.count-in-0.consumer.start-offset=earliest
spring.cloud.stream.kafka.streams.bindings.count-in-0.consumer.key-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings.count-in-0.consumer.value-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings.count-out-0.producer.key-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings.count-out-0.producer.value-serde=org.apache.kafka.common.serialization.Serdes$LongSerde准备独有 topic、确认 broker 后才运行 mvn spring-boot:run。不要在未经核对的共享 broker 上自动创建/重置已有业务资源。输入 value 为 A、a 的两条记录，预期最终 key=a 的聚合值为2；缓存可能合并中间更新，不能据“只看见2没看见1”判定丢消息。输出是 Long serde 编码，不能只用未配置Long反序列化器的字符串控制台判断错误。
applicationId 是恢复身份，不是装饰性应用名。 它关联 Kafka Streams 的 consumer group、内部 topic 与状态恢复。相同拓扑的实例用相同 applicationId，分担 stream task；不同拓扑误用同一 id，会把任务分配与状态目录混在一起。改 id 会成为独立应用身份，按无旧位点条件重新读输入并建立另一套内部状态，外部业务副作用可能重复。state.dir 是本机缓存状态目录；删除它不等于清除 broker changelog，改变 store 名/serde/拓扑也不是安全的“二进制滚回”。固定实现按函数名查函数级配置并应用 applicationId：
AbstractKafkaStreamsBinderProcessor.java:206–222（固定commit，SHA-256见元数据）：
if (kafkaStreamsBinderConfigurationProperties != null) {
final Map<String, KafkaStreamsBinderConfigurationProperties.Functions> functionConfigMap = kafkaStreamsBinderConfigurationProperties.getFunctions();
if (!CollectionUtils.isEmpty(functionConfigMap)) {
final KafkaStreamsBinderConfigurationProperties.Functions functionConfig = functionConfigMap.get(beanNamePostPrefix);
if (functionConfig != null) {
final Map<String, String> functionSpecificConfig = functionConfig.getConfiguration();
if (!CollectionUtils.isEmpty(functionSpecificConfig)) {
streamConfiguration.putAll(functionSpecificConfig);
}
String applicationId = functionConfig.getApplicationId();
if (StringUtils.hasText(applicationId)) {
streamConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
}
}
}
}普通 binder 的 ackMode=MANUAL 不能作为本拓扑逐条 ack API 使用。Kafka Streams 以 task/拓扑处理、状态更新、commit interval 和所选 processing.guarantee 管理恢复；exactly_once_v2 能约束相应 Kafka 输入、状态与输出的事务，不包含发送邮件/写外部MySQL。rebalance 后 task 重新分配，本地 store 可从 changelog 恢复，但 store 变更、权限不足、磁盘满、恢复时间过长都需独立诊断。
失败验证从三个可解释输入开始：把输出 value-serde 误配为String观察序列化错误；保持id重启再查聚合延续；改为新id后对比重放与新内部topic。先记录实际 application.id、拓扑描述、store名、线程/分区分配、最后commit与恢复日志，再决定是serde、身份还是恢复资源问题，不以“进程还活着”断言流仍在推进。故障消除后用已知输入重新确认最终计数与外部副作用不变量。
容量算例（教学假设）：100万个 key、每个聚合原始状态80B，原始状态约80MB；这不是RocksDB/RSS估算，还要计索引、缓存、changelog、恢复网络和副本。普通Consumer+数据库聚合能用数据库事务/唯一约束表达跨表业务，代价是远程写放大；Kafka Streams把key状态下沉本机，代价是分区拓扑、状态迁移及恢复运维。只有key分区、状态规模和恢复窗口符合要求时才选择后者。固定官方来源为usage及拓扑文档与上述源码。
另外，旧导航仍出现 reactive Kafka binder，但5.0.4的活动POM没有其源码模块，BOM还残留依赖坐标；其固定overview首行声明自4.3弃用并建议迁移普通Kafka binder+Reactor类型。不能把陈旧导航当成“现役可解析5.0.4构件”的证据，也不能把这种迁移解释成自动获得端到端背压。
Pulsar binder：订阅、schema 与确认差异
Pulsar binder 用相同的函数/绑定入口，但实际消费身份是 subscription，而不是 Kafka 的 __consumer_offsets 消费组。单条确认、累计确认、Shared/Exclusive等订阅行为受Pulsar客户端与broker约束；Kafka的分区commit与Rabbit的delivery tag都不能直接套用。这里讲Stream适配，Spring Pulsar直连API的完整课归U48。
本节全部是源码核查与未运行配置推演。工程仍为Boot4.0.9/Cloud2025.1.4；实际固定POM坐标为 org.springframework.cloud:spring-cloud-stream-binder-pulsar:5.0.4。旧固定文档仍写 org.springframework.pulsar:spring-pulsar-spring-cloud-stream-binder，与本tag活动POM不一致，不能盲抄。Boot管理Spring Pulsar2.0.8、Pulsar client4.1.3；本机没有Pulsar broker，未声明任意broker版本组合已验证。
独立工程只加入下列binder，并使用本页通用Boot应用入口，把函数Bean设为 Function<String,String> echo() { return value -> value; }；输入/输出分别是两个独有topic，避免自循环：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-pulsar</artifactId>
<version>5.0.4</version>
</dependency>spring.main.web-application-type=none
spring.cloud.function.definition=echo
spring.pulsar.client.service-url=pulsar://127.0.0.1:6650
spring.pulsar.administration.service-url=http://127.0.0.1:8080
spring.cloud.stream.bindings.echo-in-0.destination=persistent://public/default/u36-input
spring.cloud.stream.bindings.echo-out-0.destination=persistent://public/default/u36-output
spring.cloud.stream.bindings.echo-in-0.consumer.use-native-decoding=true
spring.cloud.stream.bindings.echo-out-0.producer.use-native-encoding=true
spring.cloud.stream.pulsar.bindings.echo-in-0.consumer.schema-type=STRING
spring.cloud.stream.pulsar.bindings.echo-in-0.consumer.message-type=java.lang.String
spring.cloud.stream.pulsar.bindings.echo-out-0.producer.schema-type=STRING
spring.cloud.stream.pulsar.bindings.echo-out-0.producer.message-type=java.lang.String
spring.cloud.stream.pulsar.bindings.echo-in-0.consumer.subscription.name=u36-echo
spring.cloud.stream.pulsar.bindings.echo-in-0.consumer.subscription.type=Shared
spring.cloud.stream.pulsar.bindings.echo-in-0.consumer.ack.timeout=30s
spring.cloud.stream.pulsar.bindings.echo-in-0.consumer.ack.redelivery-delay=5s
spring.cloud.stream.pulsar.bindings.echo-in-0.consumer.ack.group-time=100ms6650是客户端二进制协议端点，8080是管理HTTP端点，不能互换；带鉴权时两者权限都需核对。仅配置 bindings.echo-in-0.group=u36-echo 不能假设会变成同名subscription：该固定实现生成订阅名只读Pulsar扩展属性的 subscription.name，否则生成 destination-anon-subscription-UUID，createConsumerEndpoint 的group参数没有参与此处。重启生成新订阅的行为不能按Kafka显式group经验判断。实际分支：
PulsarBinderUtils.java:60–65（固定commit，SHA-256见元数据）：
static String subscriptionName(PulsarConsumerProperties consumerProps, ConsumerDestination consumerDestination) {
if (StringUtils.hasText(consumerProps.getSubscription().getName())) {
return consumerProps.getSubscription().getName();
}
return SUBSCRIPTION_NAME_FORMAT_STR.formatted(consumerDestination.getName(), UUID.randomUUID());
}默认非native转换先由Stream把对象转byte[]，Pulsar发送 Schema.BYTES；native开关打开后binder用 schema-type、message-type、必要时 message-key-type 求schema。若要JSON/AVRO对象须给真实目标Class，缺少它可能在绑定初始化确定schema时失败。换JSON schema不只是把contentType字符串改成json；生产者和消费者必须同意writer/reader类型。
PulsarMessageChannelBinder.java:146–158（固定commit，SHA-256见元数据）：
if (properties.isUseNativeDecoding()) {
var schemaType = Optional.ofNullable(properties.getExtension().getSchemaType()).orElse(SchemaType.NONE);
var schema = this.schemaResolver
.resolveSchema(schemaType, properties.getExtension().getMessageType(),
properties.getExtension().getMessageKeyType())
.orElseThrow(() -> "Could not determine consumer schema for " + destination.getName());
containerProperties.setSchema(schema);
}
else {
containerProperties.setSchema(Schema.BYTES);
}
var subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination);
containerProperties.setSubscriptionName(subscriptionName);上例 consumer.ack.* 是Pulsar客户端确认分组/超时/negative-ack重投时间参数，单位用Duration，不是Kafka ackMode。timeout=30s 是未确认后允许重投的窗口，不是函数执行截止时间，也不取消已经开始的业务副作用。代码新建 PulsarContainerProperties 并注册record listener，再把扩展 consumer属性映射到Pulsar客户端；没有从Kafka风格属性读取MANUAL确认模式，也没有在此处分发通用 Acknowledgment 头。不能仅加 spring.kafka.listener.ack-mode=manual 或照搬Kafka头来声称实现手工确认。要手工确认应选U48的对应直连API并核对Pulsar容器ack配置、消息ID及失败路径，或提供经过验证的binder定制。
误配验证（未运行预期）：管理URL错而客户端URL对时，topic provision可能先失败；native schema/class不匹配时记录绑定阶段异常；同subscription+Exclusive启动第二实例时应按broker约束查订阅冲突，改Shared再查分摊；业务处理超过未确认超时后观察同messageId重复交付，不能只看最终输出总数。恢复先修schema/权限/端点，再对同subscription核对未确认数量、重投计数与业务幂等键。这里没有证明broker持久化、复制、实际重投或DLQ，不能把Kafka实验的PASS移植给Pulsar。
选择算例：假设200条/s、单次业务40ms，平均约8个在途任务；把接收队列1000误当业务并发上限会积压约5秒输入。Shared订阅可在消费者间分摊但顺序约束需单独核对；Key_Shared等模式对key、批处理与顺序有额外条件。保留Kafka binder能复用现有位点运维与故障证据，切Pulsar需要重做subscription/schema/确认及恢复验证，成本不因函数签名相同而消失。固定来源：PulsarMessageChannelBinder、同目录properties/ConsumerConfigProperties.java，以及实际模块POM。
Schema Registry：版本登记与兼容不是一件事
Schema Registry的三个活动模块分别是：core提供模型、校验器、JPA仓库与REST控制器；server把core打包成独立Boot服务；client提供登记/查询客户端、缓存与Avro转换器。它把“消息携带一份完整schema”改为“携带能定位writer schema的引用”，减少重复载荷，并允许消费者选择reader schema。它不提供Kafka复制，不替代认证，也不自动确保业务字段变化兼容。
下面是独立服务的完整教学入口，未运行本服务。独立Boot4.0.9/Cloud2025.1.4工程加入 org.springframework.cloud:spring-cloud-stream-schema-registry-core:5.0.4，H2为该模块已有依赖；也可用同版server打包入口，不能照搬旧文档4.0.3可执行jar声称5.0.4实跑。
package study.schema;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.schema.registry.EnableSchemaRegistryServer;
@SpringBootApplication
@EnableSchemaRegistryServer
public class SchemaApp {
public static void main(String[] args) {
SpringApplication.run(SchemaApp.class, args);
}
}server.address=127.0.0.1
server.port=18091
spring.datasource.url=jdbc:h2:mem:u36schema;DB_CLOSE_DELAY=-1
spring.datasource.username=sa
spring.datasource.password=
spring.jpa.hibernate.ddl-auto=create-drop
spring.cloud.stream.schema.server.allow-schema-deletion=false此处内存H2和create-drop只用于一次性合成数据；重启会丢注册表，不能用它证明生产恢复。实际生产需要持久库、备份与权限，而不是开放删除后“有错再重建”。准备好独立环境后用 mvn spring-boot:run 启动，再调用：
curl -i -H 'Content-Type: application/json' \
-d '{"subject":"u36-value","format":"avro","definition":"\"string\""}' \
http://127.0.0.1:18091/
curl -i http://127.0.0.1:18091/u36-value/avro/v1
curl -i -H 'Content-Type: application/json' \
-d '{"subject":"u36-value","format":"avro","definition":"\"long\""}' \
http://127.0.0.1:18091/预期第一份得到id/version=1，重复登记相同定义复用已有版本，第二个合法但不同的定义成为version=2。精确路径是 /{subject}/{format}/v{version}，不是去掉v的路径。重点反例：string→long不满足相同字段的Avro reader/writer类型解析，但注册仍可能成功。控制器只validate语法、查已注册schema的等价匹配、分配新版本；默认Avro校验器的compatibilityCheck返回null，并没有把兼容策略作为注册门禁：
ServerController.java:90–104（固定commit，SHA-256见元数据）：
validator.validate(schema.getDefinition());
Schema result;
List<Schema> registeredEntities =
this.repository.findBySubjectAndFormatOrderByVersion(schema.getSubject(), schema.getFormat());
if (registeredEntities.isEmpty()) {
schema.setVersion(1);
result = this.repository.save(schema);
}
else {
result = validator.match(registeredEntities, schema.getDefinition());
if (result == null) {
schema.setVersion(registeredEntities.get(registeredEntities.size() - 1).getVersion() + 1);
result = this.repository.save(schema);
}AvroSchemaValidator.java:59–62（固定commit，SHA-256见元数据）：
@Override
public Compatibility compatibilityCheck(String source, String other) {
return null;
}client工程使用 org.springframework.cloud:spring-cloud-stream-schema-registry-client:5.0.4，公开API是 SchemaRegistryClient.register(subject,format,definition) 与 fetch(SchemaReference)/fetch(id)。例如在独立客户端 @Bean 显式配置：
@Bean
org.springframework.cloud.stream.schema.registry.client.SchemaRegistryClient schemaRegistryClient() {
var client = new org.springframework.cloud.stream.schema.registry.client.DefaultSchemaRegistryClient(
new org.springframework.web.client.RestTemplate());
client.setEndpoint("http://127.0.0.1:18091");
return client;
}或使用 @EnableSchemaRegistryClient + spring.cloud.stream.schema-registry-client.endpoint=http://127.0.0.1:18091（固定属性类前缀）。Avro converter的binding content-type=application/*+avro 选择转换器；reader schema、subject命名策略和缓存属于client层。Confluent客户端是另一种registry协议实现，不可把Spring server的路径/返回值假设为Confluent协议。无论用哪种registry，schema可取回与reader能解码该writer数据是两条独立断言。
失败输入依次检查：format=unknown→不支持格式；Avro定义语法错→校验失败；请求不存在id→找不到schema；已登记string writer、long reader→解码兼容性失败。恢复时先记录消息schema引用、实际writer/reader全文和转换器版本，修reader或数据演进策略，再回放已知载荷；只重新注册一个版本不能修复旧消息。故障中缓存命中可能继续解码，缓存未命中必须访问registry，因此“某实例没报错”也不能证明registry可用。
容量与方案选择（假设）：2KB schema×1000条/s，若每消息内联schema，约额外2MB/s；用引用与本机缓存可降低重复载荷，代价是冷启动/新版本取schema的远程依赖、缓存失效与注册表恢复。JSON自描述载荷部署简单但缺少统一reader/writer schema规则；Avro+registry适合需要约束演进的多消费者场景，但应在CI显式加入Avro兼容测试。上述服务、HTTP调用和Avro兼容解码本轮均未运行，只由固定控制流支持这些条件性预期。固定来源：Schema Registry三个模块。
FunctionCatalog 与类型转换
Spring Cloud Function 的核心是 FunctionCatalog：应用上下文里所有 Function/Consumer/Supplier 都被注册进来，调用方用定义字符串（definition）查找而不是用 bean 名。lookup(functionDefinition, expectedOutputMimeTypes...) 的契约写得很清楚：查找假设函数子类型是 Message<byte[]> 系（Function<Message<byte[]>, Message<byte[]>> 或 Flux 版本），acceptedOutputMimeTypes 用于把函数输出转换回 Message<byte[]>：
/**
* Will look up the instance of the functional interface by name only.
* This lookup method assumes a very specific semantics which are: <i>function sub-type(s)
* expected to be {@code Message<byte[]>}</i>. <br>
* For example,
* <br><br>
* {@code Function<Message<byte[]>, Message<byte[]>>} or
* <br>
* {@code Function<Flux<Message<byte[]>>, Flux<Message<byte[]>>>} or
* <br>
* {@code Consumer<Flux<Message<Flux<Message<byte[]>>>} etc. . .
* <br><br>
* The {@code acceptedOutputMimeTypes} are the string representation of {@link MimeType} where each
* mime-type in the provided array would correspond to the output with the same index
* (for cases of functions with multiple outputs) and is used to convert such output back
* to {@code Message<byte[]>}.
* If you need to provide several accepted types per specific output you can simply delimit
* them with comma (e.g., {@code application/json,text/plain...}).
*
* @param <T> instance type which should be one of {@link Supplier}, {@link Function} or {@link Consumer}.
* @param functionDefinition the definition of a function (e.g., 'foo' or 'foo|bar')
* @param acceptedOutputMimeTypes acceptedOutputMimeTypes array of string representation of {@link MimeType}s
* used to convert function output back to {@code Message<byte[]>}.
* @return instance of the functional interface registered with this catalog
*/
default <T> T lookup(String functionDefinition, String... expectedOutputMimeTypes) {
return this.lookup(null, functionDefinition, expectedOutputMimeTypes);
}
<T> T lookup(Class<?> type, String functionDefinition, String... expectedOutputMimeTypes);SimpleFunctionRegistry.lookup 先做定义规范化、再进入真正查找；找不到返回 null 而不是抛异常（debug 日志 “Failed to locate function”）：
@SuppressWarnings("unchecked")
@Override
public <T> T lookup(Class<?> type, String functionDefinition, String... expectedOutputMimeTypes) {
functionDefinition = this.normalizeFunctionDefinition(functionDefinition);
FunctionInvocationWrapper function = this.doLookup(type, functionDefinition, expectedOutputMimeTypes);
if (logger.isDebugEnabled()) {
if (function != null) {
logger.debug("Located function: " + function);
}
else {
logger.debug("Failed to locate function: " + functionDefinition);
}
}
return (T) function;
}规范化规则是 Function 的一个高频坑：逗号被替换为 |；若 catalog 里只有一个函数且名字不匹配，会回退到这个唯一函数（单个函数回退，FunctionProperties.singleFunctionFallbackEnabled 默认开启，排除 functionRouter 与 defaultMessageRoutingHandler）：
/**
* This method will make sure that if there is only one function in catalog
* it can be looked up by any name or no name.
* It does so by attempting to determine the default function name
* (the only function in catalog) and checking if it matches the provided name
* replacing it if it does not.
*/
String normalizeFunctionDefinition(String functionDefinition) {
functionDefinition = StringUtils.hasText(functionDefinition)
? functionDefinition.replaceAll(",", "|")
: System.getProperty(FunctionProperties.FUNCTION_DEFINITION, "");
Set<String> names = this.getNames(null);
if (this.isSingleFunctionFallbackEnabled() && !names.contains(functionDefinition)) {
List<String> eligibleFunction = names.stream()
.filter(name -> !RoutingFunction.FUNCTION_NAME.equals(name))
.filter(name -> !RoutingFunction.DEFAULT_ROUTE_HANDLER.equals(name))
.collect(Collectors.toList());
if (eligibleFunction.size() == 1
&& !eligibleFunction.get(0).equals(functionDefinition)
&& !functionDefinition.contains("|")
&& !eligibleFunction.get(0).startsWith("&")) {
functionDefinition = eligibleFunction.get(0);
}
}
return functionDefinition;
}
private boolean isSingleFunctionFallbackEnabled() {
return this.functionProperties == null || this.functionProperties.isSingleFunctionFallbackEnabled();
}这意味着“definition 写错也能跑”在单函数应用里是事实——多函数应用才会暴露拼写错误（返回 null，绑定/路由失败）。类型转换发生在调用边界：registry 持有 ConversionService 与 CompositeMessageConverter，FunctionInvocationWrapper 在调用前后按输入/输出类型与 MIME 转换。把 catalog lookup 当成“bean 查找”会误判失败形态：bean 查找失败抛 NoSuchBeanDefinitionException，catalog 查找失败是 null。
函数组合、路由与响应式函数
组合：spring.cloud.function.definition=uppercase|exclaim 把两个函数按顺序 andThen 链成一个。compose 的实现逐个查找函数、用 FunctionInvocationWrapper.andThen 串联、并在每步补充输入/输出的类型丰富（enrich），最后把单例组合缓存在 wrappedFunctionDefinitions：
private FunctionInvocationWrapper compose(Class<?> type, String functionDefinition) {
String[] functionNames = StringUtils.delimitedListToStringArray(functionDefinition.replaceAll(",", "|").trim(), "|");
FunctionInvocationWrapper composedFunction = null;
for (String functionName : functionNames) {
FunctionInvocationWrapper function = this.findFunctionInFunctionRegistrations(functionName);
if (function == null) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to locate function '" + functionName + "' for function definition '"
+ functionDefinition + " amongst existing function registrations. Will check with the bean factory");
}
return null;
}
else {
if (composedFunction == null) {
composedFunction = function;
}
else {
FunctionInvocationWrapper andThenFunction =
invocationWrapperInstance(functionName, function.getTarget(), function.inputType, function.outputType);
composedFunction = (FunctionInvocationWrapper) composedFunction.andThen((Function<Object, Object>) andThenFunction);
}
composedFunction = this.enrichInputIfNecessary(composedFunction);
composedFunction = this.enrichOutputIfNecessary(composedFunction);
}
}
if (composedFunction != null && composedFunction.isSingleton) {
synchronized (cacheLock) {
this.wrappedFunctionDefinitions.put(composedFunction.functionDefinition, composedFunction);
}
}
if (logger.isDebugEnabled()) {
logger.debug("Composed function " + composedFunction);
}
return composedFunction;
}组合是单管道顺序执行，不是并行编排，也不是消息分发；Consumer 只能出现在链尾、Supplier 只能出现在链首，类型不匹配在组合查找阶段就失败（返回 null 或抛错）。
路由：RoutingFunction（bean 名 functionRouter）先尝试应用的 MessageRoutingCallback；没有给出结果时，再看消息头 spring.cloud.function.definition，其后为消息头 spring.cloud.function.routing-expression，然后是应用级 routing-expression、应用级 definition（RoutingFunction.java:132-161,196-220）。header 表达式使用只读 SimpleEvaluationContext，应用表达式使用 StandardEvaluationContext；只读求值不等于目标函数已获授权。外部入口必须剥离未授权路由头，并用可信 callback 把允许的业务操作映射到白名单函数；不要把用户输入拼成应用级 SpEL。
响应式函数适配
响应式：函数签名可以是 Function<Flux<String>, Flux<String>>。registry 用 isTypePublisher 判定输入/输出是否为 Publisher；当输入类型不是 Publisher 但实际收到 Publisher 时，逐元素 map 调用再重组流，并对结果流挂 doOnError 日志：
private Object invokeFunction(Object convertedInput) {
Object result;
if (!this.isTypePublisher(this.inputType) && convertedInput instanceof Publisher publisherInput) {
result = publisherInput instanceof Mono
? Mono.from(publisherInput).map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value))
.doOnError(ex -> logger.error("Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex))
: Flux.from(publisherInput).map(value -> this.invokeFunctionAndEnrichResultIfNecessary(value))
.doOnError(ex -> logger.error("Failed to invoke function '" + this.functionDefinition + "'", (Throwable) ex));
}
else {
result = this.invokeFunctionAndEnrichResultIfNecessary(convertedInput);
if (result instanceof Flux flux) {
result = flux.doOnError(ex -> logger.error("Failed to invoke function '"
+ this.functionDefinition + "'", (Throwable) ex));
}
else if (result instanceof Mono mono) {
result = mono.doOnError(ex -> logger.error("Failed to invoke function '"
+ this.functionDefinition + "'", (Throwable) ex));
}
}
return result;
}两个边界必须记：其一，响应式函数不是命令式 Message handler，流内错误不经命令式 RetryTemplate/DLQ，需用 Reactor 错误算子和明确的恢复策略；其二，Stream 在绑定阶段把输入 channel 转成 Flux，调用一次 functionToInvoke.apply(...) 建立函数管道，再订阅输出，并非对每个 broker record 重新调用整个 Flux 函数。普通 Kafka binder 底层仍使用阻塞 Kafka 容器，不能因此声称端到端背压或把流内成功直接当成每条记录安全提交。见 FunctionConfiguration.java:509-580,637 与 Reactor核心。
原码分别展示输入channel转Flux、一次构建管道，以及路由来源的真实判断顺序：
Publisher[] inputPublishers = inputBindingNames.stream().map(inputBindingName -> {
BindingProperties bindingProperties = this.serviceProperties.getBindings().get(inputBindingName);
ConsumerProperties consumerProperties = bindingProperties == null ? null : bindingProperties.getConsumer();
if (consumerProperties != null) {
function.setSkipInputConversion(consumerProperties.isUseNativeDecoding());
if (consumerProperties.getConcurrency() > 1) {
this.logger.warn("When using concurrency > 1 in reactive contexts, please make sure that you are using a " +
"reactive binder that supports concurrency settings. Otherwise, concurrency settings > 1 will be ignored when " +
"using reactive types.");
}
}
MessageChannel inputChannel = this.applicationContext.getBean(inputBindingName, MessageChannel.class);
return IntegrationReactiveUtils.messageChannelToFlux(inputChannel).map(m -> {
if (m != null) {
m = sanitize(m);
}
return m;
}); }
functionToInvoke = new PartitionAwareFunctionWrapper(function, this.applicationContext, producerProperties);
// If we have a multi-output scenario, we will do any message enrichment (aka, determining the outbound
// partition) via the corresponding reactive Flux types. Currently, we support multiple output
// bindings for reactive types only (Tuples).
if (outputBindingNames.size() > 1) {
((PartitionAwareFunctionWrapper) functionToInvoke).setMessageEnricherEnabled(false);
}
}
Object resultPublishers = functionToInvoke.apply(inputPublishers.length == 1 ? inputPublishers[0] : Tuples.fromArray(inputPublishers)); if (this.routingCallback != null) {
String functionDefinition = this.routingCallback.routingResult(message);
if (StringUtils.hasText(functionDefinition)) {
function = this.functionFromDefinition(functionDefinition);
}
}
if (function == null) {
function = this.locateFunctionFromDefinitionOrExpression(message);
if (function != null) {
if (function.isInputTypePublisher()) {
this.assertOriginalInputIsNotPublisher(originalInputIsPublisher);
}
}
else if (StringUtils.hasText(functionProperties.getRoutingExpression())) {
function = this.functionFromExpression(functionProperties.getRoutingExpression(), message);
}
else if (StringUtils.hasText(functionProperties.getDefinition())) {
function = this.functionFromDefinition(functionProperties.getDefinition());
}
else {
throw new IllegalStateException("Failed to establish route, since neither were provided: "
+ "'spring.cloud.function.definition' as Message header or as application property or "
+ "'spring.cloud.function.routing-expression' as application property. Incoming message: " + input);
}
}
}
else if (input instanceof Publisher<?> publisher) {
if (StringUtils.hasText(functionProperties.getDefinition())) {
function = functionFromDefinition(functionProperties.getDefinition());
} private FunctionInvocationWrapper locateFunctionFromDefinitionOrExpression(Message<?> message) {
Map<String, Object> messageHeaders = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
messageHeaders.putAll(message.getHeaders());
Object defValue = messageHeaders.get(FunctionProperties.FUNCTION_DEFINITION);
Object exprValue = messageHeaders.get(FunctionProperties.ROUTING_EXPRESSION);
if (defValue != null) {
if (defValue instanceof String definition) {
return functionFromDefinition(definition);
}
else if (defValue instanceof List<?> definitions && !definitions.isEmpty()) {
return functionFromDefinition(definitions.stream().map(Object::toString).collect(Collectors.joining(",")));
}
}
else if (exprValue != null) {
if (exprValue instanceof String expression) {
return functionFromExpression(expression, message, true);
}
else if (exprValue instanceof List<?> expressions && !expressions.isEmpty()) {
return functionFromExpression(expressions.get(0).toString(), message, true);
}
}
return null;
}独立 catalog 的调用与可信路由
只使用 catalog 时可以完全不接 broker：新建独立目录，沿用前文完整 POM 的 parent/BOM/插件，将 dependencies 改为 spring-boot-starter 和 spring-cloud-function-context，不复制 Stream 的 application.yml。以下 src/main/java/study/catalog/CatalogDemo.java 同时给出普通组合、响应式输入及白名单路由：
package study.catalog;
import java.time.Duration;
import java.util.Locale;
import java.util.function.Function;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.MessageRoutingCallback;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import reactor.core.publisher.Flux;
@SpringBootApplication
public class CatalogDemo {
public static void main(String[] args) {
try (var context = new SpringApplicationBuilder(CatalogDemo.class)
.web(WebApplicationType.NONE).run(args)) {
FunctionCatalog catalog = context.getBean(FunctionCatalog.class);
Function<String, String> combined = catalog.lookup("uppercase|exclaim");
System.out.println(combined.apply("hello")); // 预期 HELLO!
Function<Flux<String>, Flux<String>> stream = catalog.lookup("streamUpper");
System.out.println(stream.apply(Flux.just("a", "b"))
.collectList().block(Duration.ofSeconds(2))); // 预期 [A, B]
Function<Object, Object> router = catalog.lookup("functionRouter");
Object routed = router.apply(MessageBuilder.withPayload("hello")
.setHeader("operation", "loud").build());
Object payload = routed instanceof Message<?> message ? message.getPayload() : routed;
System.out.println(payload); // 此调用的输入是 Message，输出也可能携带 Message 包装。
}
}
@Bean
public Function<String, String> uppercase() {
return value -> value.toUpperCase(Locale.ROOT);
}
@Bean
public Function<String, String> exclaim() { return value -> value + "!"; }
@Bean
public Function<Flux<String>, Flux<String>> streamUpper() {
return values -> values.map(value -> value.toUpperCase(Locale.ROOT));
}
@Bean
public MessageRoutingCallback trustedRoutes() {
return new MessageRoutingCallback() {
@Override
public String routingResult(Message<?> message) {
String operation = message.getHeaders().get("operation", String.class);
return switch (operation == null ? "" : operation) {
case "upper" -> "uppercase";
case "loud" -> "uppercase|exclaim";
default -> throw new IllegalArgumentException("operation not allowed");
};
}
};
}
}执行 mvn spring-boot:run 的输入是类中三个合成调用；输出注释为预期值，稳定的本次实测结果以公开 runner 的证据为准。这个 callback 对 Message 输入优先于 definition/表达式来源，且始终返回固定白名单或抛错，不在拒绝分支返回 null 让框架继续回退。把 operation=deleteAll 或缺失 operation 作为负例，应在进入业务函数前拒绝；这比把外部 header 直接当成函数 definition 清晰。它仍不是鉴权：真实入口必须先确认身份、租户和该身份允许的 operation，并剥离外部提供的 spring.cloud.function.* 路由头；只读 SpEL 上下文也不等于业务授权。
把 streamUpper 改成 values -> values.map(v -> { throw ...; })，错误发生在订阅后的流内；围绕 catalog.lookup 写 try/catch 捕获不到该异步信号。应在返回的流中定义错误恢复或传播，核对订阅是否结束及资源是否释放。retryWhen 可能重新订阅整个源，它不会自动提供 per-record 的业务幂等；不能把 catalog 流示例的结果外推成 Kafka 中的恢复/确认协议。
Function web 适配
spring-cloud-function-web 把函数暴露为 HTTP：路径默认等于函数名，FunctionHandlerMapping（RequestMappingHandlerMapping 子类）拦截请求、用 FunctionWebRequestProcessingHelper 按 HTTP 方法与路径找函数，找到就把函数放进请求属性交给 FunctionController 调用；spring.cloud.function.web.path 可加前缀（spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/mvc/FunctionHandlerMapping.java:46-108）。含义：/uppercase 走函数 uppercase，Accept/Content-Type 参与 MIME 选择与转换（不支持的媒体类型与函数查找失败属于不同路径；无单函数或应用 definition 回退且找不到目标时才进入未匹配路径，不能把未知 URL 一律写成404，也不能把所有转换异常一律写成415），functionRouter 经 header 表达式可做路径级路由。它不是通用 MVC：没有 controller 注解、没有方法级参数绑定，只有“函数名即路径”的约定。
本轮 Function web 未运行：配置与 HTTP 映射为固定源码推演；本票只运行 catalog 直调。单函数回退、应用 definition、不同 MIME 的真实 HTTP 状态仍需对应 web adapter 专项测试，不能用编译通过代替。普通 Kafka binder 接 Flux 的装配机制已按固定源码核对，但公开实验没有覆盖响应式 binder 的提交与错误恢复。
Function web 的独立配置
此配置只用于理解 web adapter，未运行。在 catalog 独立项目中添加 BOM 管理的 spring-cloud-function-web 与 Boot 管理的 spring-boot-starter-webmvc；把 CatalogDemo.main 改为 org.springframework.boot.SpringApplication.run(CatalogDemo.class, args)，使应用保留 HTTP 服务，不再执行 main 中的三次直调并立即关闭。用以下配置固定本机监听与路径：
server:
address: 127.0.0.1
port: 0
spring:
cloud:
function:
web:
path: /fn启动后读取实际分配端口，对 /fn/uppercase 发 POST，Content-Type: text/plain、Accept: text/plain、body=hello，预期返回函数结果；访问不存在函数的路径与使用不支持的媒体类型分别检查路由/转换失败。这里的 HTTP 状态须以所选 adapter 的真实响应复核，不把“转换失败”统一当作同一个状态码。改 web.path 只改变入口映射，不给函数加安全过滤链或调用预算；若开放路由函数，同样要执行上一节的身份授权与路由白名单。
诊断顺序是实际监听端口 → 前缀/函数 definition → MIME 转换 → 函数业务异常。把 /uppercase 与配置后的 /fn/uppercase 混用，是路由错误，不是 Kafka 不通；把 Content-Type 改成 JSON 却仍发送未加引号的字符串，则是载荷/媒体类型不匹配。修复后用同一输入、正确路径和正确头重试，并区分“函数返回”和“HTTP 响应已写出”两个时点。
Function adapters：同一函数的传输、装载与流程接点
FunctionCatalog 可以复用 uppercase 的业务代码，adapter决定谁创建ApplicationContext、输入怎样变为Message、输出怎样映射到宿主响应。AWS、Azure、GCP、gRPC及serverless-web是5.0.5活动源码模块，不能因为云账号未配置就从知识清单删掉。此处覆盖Java入口与机制；创建云资源、IAM、打包上传、冷启动和收费等厂商交付流程按PRD专有平台边界排除。以下均为固定源码核查和未运行的教学样例；公开实验只验证Catalog行为，没有验证任何云部署。
建立独立Boot4.0.9/Cloud2025.1.4工程，先用共同入口 src/main/java/study/adapters/AdapterApp.java：
package study.adapters;
import java.util.Locale;
import java.util.function.Function;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class AdapterApp {
public static void main(String[] args) {
SpringApplication.run(AdapterApp.class, args);
}
@Bean
Function<String, String> uppercase() {
return value -> value.toUpperCase(Locale.ROOT);
}
}各适配工程只选自己的artifact。所有下表坐标的groupId为 org.springframework.cloud、version为 5.0.5，使用相同依赖写法，例如AWS：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-adapter-aws</artifactId>
<version>5.0.5</version>
</dependency>spring.main.web-application-type=none
spring.cloud.function.definition=uppercase| 活动模块 | Java接点及配置 | 能复用与必须重新验证的部分 |
|---|---|---|
spring-cloud-function-adapter-aws | handler设为 org.springframework.cloud.function.adapter.aws.FunctionInvoker；宿主环境变量可设 spring_cloud_function_definition=uppercase | 实现 RequestStreamHandler.handleRequest(InputStream,OutputStream,Context)；适配事件/JSON并查Catalog。Lambda重试、API Gateway envelope、超时和IAM属于宿主；不是函数签名能证明的能力 |
spring-cloud-function-adapter-gcp | 打包Boot应用的entry point是 org.springframework.cloud.function.adapter.gcp.GcfJarLauncher；保持上述函数definition | launcher创建Boot依赖classloader并委托FunctionInvoker，分别实现HTTP service和RawBackgroundFunction accept；不能把AWS handler类填入GCP配置 |
spring-cloud-function-adapter-azure | Java Worker通过SPI调用 AzureFunctionInstanceInjector.getInstance；下面的@Component触发器桥接到Catalog | Worker仍识别Azure的 @FunctionName/@HttpTrigger；Spring注入、函数计算可复用，Azure trigger/binding/重投配置仍由宿主定义 |
spring-cloud-function-serverless-web | ServerlessMVC.INSTANCE(AdapterWebApp.class) 后以 service(HttpServletRequest,HttpServletResponse) 转发 | 初始化DispatcherServlet与过滤链，使用轻量Servlet代理；不等于完整Servlet容器在任意异步、session、streaming场景的保证 |
spring-cloud-function-adapter-azure-web | Azure Worker到 AzureWebProxyInvoker，后者缓存并调用ServerlessMVC；是上一行在Azure的HTTP代理 | 与普通Azure Bean注入适配不同，目标是MVC页面/Controller。认证、路径映射、状态头、二进制体、超时都要实际代理请求验证 |
spring-cloud-function-grpc | spring.cloud.function.grpc.server=true、port=16048，复用共同入口；可设 service-class-name 选择要暴露的具体BindableService实现类 | GrpcSpringMessage 是protobuf的bytes payload+string headers；支持requestReply/serverStream/clientStream/biStream。四种RPC的函数输入/输出基数不同，不能把Flux随意塞入unary调用 |
spring-cloud-function-grpc-cloudevent-ext | 依赖加入后由CloudEventGrpcAutoConfiguration提供CloudEvent协议的handler与converter；按具体handler类选择服务 | CloudEvent必需id/source/specVersion/type及protobuf结构独立于普通JSON HTTP。类型、转换、deadline/cancel和双向流需协议测试 |
Azure独立工程的完整桥接类如下；azure-functions-java-library 随adapter依赖提供。故意用FunctionCatalog查找，以保留组合/转换包装；路由名是常量，不接受任意外部header选择内部函数。
package study.adapters;
import java.util.Optional;
import java.util.function.Function;
import com.microsoft.azure.functions.AuthorizationLevel;
import com.microsoft.azure.functions.HttpMethod;
import com.microsoft.azure.functions.HttpRequestMessage;
import com.microsoft.azure.functions.HttpResponseMessage;
import com.microsoft.azure.functions.HttpStatus;
import com.microsoft.azure.functions.annotation.FunctionName;
import com.microsoft.azure.functions.annotation.HttpTrigger;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.stereotype.Component;
@Component
public class AzureUppercase {
private final FunctionCatalog catalog;
public AzureUppercase(FunctionCatalog catalog) {
this.catalog = catalog;
}
@FunctionName("uppercase")
public HttpResponseMessage invoke(
@HttpTrigger(name = "request", methods = HttpMethod.POST,
authLevel = AuthorizationLevel.FUNCTION)
HttpRequestMessage<Optional<String>> request) {
Function<String, String> target = catalog.lookup("uppercase");
if (target == null) {
throw new IllegalStateException("uppercase is absent");
}
String result = target.apply(request.getBody().orElse(""));
return request.createResponseBuilder(HttpStatus.OK).body(result).build();
}
}serverless-web独立工程不设 spring.main.web-application-type=none。最小MVC目标可用下列类；宿主调用 ServerlessMVC.INSTANCE(AdapterWebApp.class) 初始化，再把真实Servlet request/response传入service，不能在此伪造一场真实云请求成功：
package study.serverless;
import java.util.Locale;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
@SpringBootApplication
public class AdapterWebApp {
@RestController
static class Routes {
@PostMapping("/uppercase")
String uppercase(@RequestBody String value) {
return value.toUpperCase(Locale.ROOT);
}
}
}这里省掉 @Component 的失败不是“Azure找不到函数definition”：Worker已找到触发器类，Spring SPI却拿不到该类型的Bean。固定分支直接查 getBeansOfType(functionClass) 并在空集合时抛错：
AzureFunctionInstanceInjector.java:60–72（固定commit，SHA-256见元数据）：
@Override
public <T> T getInstance(Class<T> functionClass) throws Exception {
try {
initialize();
Map<String, T> azureFunctionBean = APPLICATION_CONTEXT.getBeansOfType(functionClass);
if (CollectionUtils.isEmpty(azureFunctionBean)) {
throw new IllegalStateException(
"Failed to retrieve Bean instance for: " + functionClass
+ ". The class should be annotated with @Component to let the Spring framework initialize it!");
}
return azureFunctionBean.entrySet().iterator().next().getValue();AWS/GCP还需满足各自可加载的归档布局与入口，不能只看到 mvn package 成功就声称云端可调用。GCP专门分开launcher与Invoker的原因可从下面几行看见：
GcfJarLauncher.java:40–59（固定commit，SHA-256见元数据）：
public GcfJarLauncher() throws Exception {
//JarFile.registerUrlProtocolHandler();
this.loader = createClassLoader(getClassPathUrls());
Class<?> clazz = this.loader
.loadClass("org.springframework.cloud.function.adapter.gcp.FunctionInvoker");
this.delegate = clazz.getConstructor().newInstance();
}
@Override
public void service(HttpRequest httpRequest, HttpResponse httpResponse) throws Exception {
Thread.currentThread().setContextClassLoader(this.loader);
((HttpFunction) delegate).service(httpRequest, httpResponse);
}
@Override
public void accept(String json, Context context) throws Exception {
Thread.currentThread().setContextClassLoader(this.loader);
((RawBackgroundFunction) delegate).accept(json, context);gRPC独立工程把依赖artifact换为 spring-cloud-function-grpc，保留 uppercase 并追加：
spring.cloud.function.grpc.server=true
spring.cloud.function.grpc.port=16048准备相同依赖的客户端后，可构造 GrpcSpringMessage.newBuilder().setPayload(ByteString.copyFromUtf8("hello")).putHeaders("contentType", "text/plain").build()，经 MessagingServiceGrpc.newBlockingStub(channel).withDeadlineAfter(2, TimeUnit.SECONDS).requestReply(request) 发送；channel 用 ManagedChannelBuilder.forAddress("127.0.0.1",16048).usePlaintext().build() 在隔离本机验证，finally中 shutdownNow()。此处是明确的protobuf调用形状，未运行远程RPC，预期payload为HELLO。生产需TLS和调用方身份。把Flux输出绑定到requestReply、把CloudEvent协议发到普通MessagingService、或不给有歧义的多函数选择定义，都是协议/基数/路由故障，不能只调大超时。固定源码中的grpc交互测试类整体有 @Disabled，不能援引源码里的测试名称当成本轮或上游CI已验证。
spring-cloud-function-integration 5.0.5 则把Catalog接入Spring Integration，并不需要broker。单独工程加入该依赖并保留Boot通用入口，将应用类改为下面的完整本机流程，spring.main.web-application-type=none：
package study.flow;
import java.util.List;
import java.util.Locale;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.function.integration.dsl.FunctionFlowBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.Pollers;
@SpringBootApplication
public class FlowApp {
public static void main(String[] args) {
SpringApplication.run(FlowApp.class, args);
}
@Bean
Supplier<String> tick() {
var iterator = List.of("alpha", "beta").iterator();
return () -> iterator.hasNext() ? iterator.next() : null;
}
@Bean
Function<String, String> uppercase() {
return value -> value.toUpperCase(Locale.ROOT);
}
@Bean
Consumer<String> sink() {
return System.out::println;
}
@Bean
IntegrationFlow flow(FunctionFlowBuilder flows) {
return flows.fromSupplier("tick", endpoint -> endpoint.poller(
Pollers.fixedDelay(1000).maxMessagesPerPoll(1)))
.apply("uppercase")
.accept("sink");
}
}此流程轮询两次，预期打印ALPHA/BETA；之后Supplier返回null表示本次poll无消息，进程仍运行，完成观察后关闭自己启动的进程。它没有承诺Task生命周期自动结束，也没经过Kafka持久化。apply是转换端点，accept是终点Consumer；已有MessageChannel可用 from(channel)接入再使用标准Integration operators。uppercase|exclaim是Catalog组合，Integration DSL则能在函数间插入channel、filter等端点，线程/错误通道/事务边界归U47深入。源码按函数名延迟lookup并缓存，故 apply("uppercas") 可能到第一条消息才失败，启动成功不是全部函数名已通过校验：
FunctionFlowDefinition.java:60–81（固定commit，SHA-256见元数据）：
public FunctionFlowDefinition apply(String functionDefinition) {
return apply(this.functionLookupHelper.lookupFunction(functionDefinition));
}
/**
* Configure a {@link Function} as a handler in the endpoint.
* @param function the {@link Function} to use.
* @return the current flow builder.
*/
public FunctionFlowDefinition apply(Function<Message<?>, ?> function) {
return handle(Message.class, (message, headers) -> function.apply(message));
}
/**
* Configure a {@link org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper}
* as a one-way handler in the final endpoint by its definition from the
* {@link org.springframework.cloud.function.context.FunctionCatalog}.
* @param consumerDefinition the consumer definition in the function catalog.
* @return the current flow builder.
*/
public IntegrationFlow accept(String consumerDefinition) {
return accept(this.functionLookupHelper.lookupConsumer(consumerDefinition));FunctionLookupHelper.java:59–66（固定commit，SHA-256见元数据）：
private <T> Supplier<T> lazyLookup(Class<?> functionType, String functionDefinition) {
return memoize(() -> requireNonNull(functionType, functionDefinition));
}
private <T> T requireNonNull(Class<?> functionType, String functionDefinition) {
T function = this.functionCatalog.lookup(functionType, functionDefinition);
Assert.notNull(function, () -> "No '" + functionDefinition + "' in the catalog");
return function;Kotlin作者也可返回标准JDK SAM类型，例如 @Bean fun uppercase(): java.util.function.Function<String,String> = java.util.function.Function { it.uppercase(java.util.Locale.ROOT) }，用同一Catalog与adapter；实际Kotlin编译插件、stdlib/reflection按U28配置。不要加入 spring-cloud-function-kotlin 作为生产依赖：固定README明确是内部Kotlin测试模块，自3.1.3起不应显式依赖；函数null返回也不会因此获得Message null payload许可。Kotlin样例属于语言互操作证据，未独立运行。旧Function deployer/compiler文档仍残留，但本tag活动POM没有相应模块，deployer行已注释，不能沿旧文档声称现役 spring.cloud.function.location 动态部署已经支持。
选择先看边界：只有单一HTTP服务且需要标准状态码/认证/资源模型时，Controller/Service通常更直接；需同一纯计算函数在消息、HTTP与指定云宿主间复用时，Catalog能减少业务重复，代价是每种传输的事件envelope、错误和生命周期仍需验证。容量算例（假设）：宿主单实例并发10、每次200ms，仅理想处理上限50次/s；多开Bean不等于提高平台并发，也不降低冷启动。故障排查记录adapter版本、归档入口、function definition、目标泛型、contentType、宿主request/event id和真实响应/输出，而非只有“Bean存在”的日志。
固定来源：Function adapters、Integration DSL、Kotlin内部模块声明。
Task 生命周期与执行记录
Spring Cloud Task 解决的是一次性短生命周期应用的簿记：启动时向 TASK_EXECUTION 表写一行（开始时间、参数、外部执行 id），结束时更新退出码、结束时间、退出消息与错误栈。TaskExecution 的字段就是这行的 Java 形态：executionId、parentExecutionId、exitCode、taskName、startTime、endTime、exitMessage、externalExecutionId、errorMessage、arguments（spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/TaskExecution.java:33-87）。
监听器与退出码
启动路径 TaskLifecycleListener.doTaskStart：若指定了 spring.cloud.task.executionid（由外部启动方预建记录），复用并校验（不存在/已结束都断言失败），否则新建 TaskExecution 并 createTaskExecution 落库；随后依次调 onTaskStartup 监听器：
private void doTaskStart() {
try {
if (!this.started) {
this.taskExecutionListeners = new ArrayList<>();
this.taskListenerExecutorObjectFactory.getObject();
if (!CollectionUtils.isEmpty(this.taskExecutionListenersFromContext)) {
this.taskExecutionListeners.addAll(this.taskExecutionListenersFromContext);
}
this.taskExecutionListeners.add(this.taskListenerExecutorObjectFactory.getObject());
List<String> args = new ArrayList<>(0);
if (this.applicationArguments != null) {
args = Arrays.asList(this.applicationArguments.getSourceArgs());
}
if (this.taskProperties.getExecutionid() != null) {
TaskExecution taskExecution = this.taskExplorer
.getTaskExecution(this.taskProperties.getExecutionid());
Assert.notNull(taskExecution, String.format("Invalid TaskExecution, ID %s not found",
this.taskProperties.getExecutionid()));
Assert.isNull(taskExecution.getEndTime(),
String.format("Invalid TaskExecution, ID %s task is already complete",
this.taskProperties.getExecutionid()));
LocalDateTime startDate = (taskExecution.getStartTime() == null) ? LocalDateTime.now()
: taskExecution.getStartTime();
this.taskExecution = this.taskRepository.startTaskExecution(this.taskProperties.getExecutionid(),
this.taskNameResolver.getTaskName(), startDate, args,
this.taskProperties.getExternalExecutionId(), this.taskProperties.getParentExecutionId());
}
else {
TaskExecution taskExecution = new TaskExecution();
taskExecution.setTaskName(this.taskNameResolver.getTaskName());
taskExecution.setStartTime(LocalDateTime.now());
taskExecution.setArguments(args);
taskExecution.setExternalExecutionId(this.taskProperties.getExternalExecutionId());
taskExecution.setParentExecutionId(this.taskProperties.getParentExecutionId());
this.taskExecution = this.taskRepository.createTaskExecution(taskExecution);
}
}
else {
logger.error("Multiple start events have been received. The first one was " + "recorded.");
}
setExitMessage(invokeOnTaskStartup(this.taskExecution));结束路径 doTaskEnd：在已启动或监听失败、且尚未结束时，设置结束时间、异常栈和退出码，随后执行失败/结束回调使其补充退出消息，最后调用 completeTaskExecution 写库，再可选关闭上下文。结束回调发生在结束记录落库之前：
private void doTaskEnd() {
if ((this.listenerFailed || this.started) && !this.finished) {
this.taskExecution.setEndTime(LocalDateTime.now());
if (this.applicationFailedException != null) {
this.taskExecution.setErrorMessage(stackTraceToString(this.applicationFailedException));
}
this.taskExecution.setExitCode(calcExitStatus());
if (this.applicationFailedException != null) {
setExitMessage(invokeOnTaskError(this.taskExecution, this.applicationFailedException));
}
setExitMessage(invokeOnTaskEnd(this.taskExecution));
this.taskRepository.completeTaskExecution(this.taskExecution.getExecutionId(),
this.taskExecution.getExitCode(), this.taskExecution.getEndTime(),
this.taskExecution.getExitMessage(), this.taskExecution.getErrorMessage());
this.finished = true;
if (this.taskProperties.getClosecontextEnabled() && this.context.isActive()) {
this.context.close();
}
}
else if (!this.started) {
logger.error("An event to end a task has been received for a task that has " + "not yet started.");
}
}退出码来源（TaskLifecycleListener.java:228-254）：已有 ExitCodeEvent 时优先使用其值；否则看监听失败异常（解包 InvocationTargetException），若实现 ExitCodeGenerator 则取其值，再回退到应用失败1、正常0。不能泛化为 Task 会自行读取任意异常链上的退出码。DB 中记录的退出码还不自动等同 OS 进程退出码，后者需要启动器显式传播。onTaskStartup 在新记录创建后调用；应用失败时 onTaskFailed 先于 onTaskEnd，这两个结束回调都在 completeTaskExecution 前：
public interface TaskExecutionListener {
/**
* Invoked after the {@link TaskExecution} has been stored in the
* {@link TaskRepository}.
* @param taskExecution instance containing the information about the current task.
*/
default void onTaskStartup(TaskExecution taskExecution) {
}
/**
* Invoked before the {@link TaskExecution} has been updated in the
* {@link TaskRepository} upon task end.
* @param taskExecution instance containing the information about the current task.
*/
default void onTaskEnd(TaskExecution taskExecution) {
}
/**
* Invoked if an uncaught exception occurs during a task execution. This invocation
* will occur before the {@link TaskExecution} has been updated in the
* {@link TaskRepository} and before the onTaskEnd is called.
* @param taskExecution instance containing the information about the current task.
* @param throwable the uncaught exception that was thrown during task execution.
*/
default void onTaskFailed(TaskExecution taskExecution, Throwable throwable) {
}
}Task 不负责触发与恢复：没有调度器，重启不自动续跑，失败不自动重试。CI、Kubernetes Job 或调度器负责启动进程，Task 记录这次执行的身份与结果。SIGKILL、进程崩溃或结束写库失败可能留下只有 start、没有 end 的记录；SIGTERM 可走优雅关闭与 stop 兜底，不能笼统说“kill 一定无 end”。对账应同时确认进程与业务状态，再决定补偿或新建执行；公开实验没有注入进程崩溃。
退出码优先级按calcExitStatus的分支核对：
private int calcExitStatus() {
int exitCode = 0;
if (this.exitCodeEvent != null) {
exitCode = this.exitCodeEvent.getExitCode();
}
else if (this.listenerFailed || this.applicationFailedException != null) {
Throwable exception = this.listenerException;
if (exception instanceof TaskExecutionException) {
TaskExecutionException taskExecutionException = (TaskExecutionException) exception;
if (taskExecutionException.getCause() instanceof InvocationTargetException) {
InvocationTargetException invocationTargetException = (InvocationTargetException) taskExecutionException
.getCause();
if (invocationTargetException != null && invocationTargetException.getTargetException() != null) {
exception = invocationTargetException.getTargetException();
}
}
}
if (exception instanceof ExitCodeGenerator) {
exitCode = ((ExitCodeGenerator) exception).getExitCode();
}
else {
exitCode = 1;
}
}
return exitCode;TaskRepository 与 schema、单实例锁
TaskRepository 是簿记的持久化 SPI：createTaskExecution（三种重载：完整对象、只给名字、只给 id——后两者服务于“先创建再启动”的分离式启动）、startTaskExecution（更新开始信息）、completeTaskExecution（写结束信息）、updateExternalExecutionId。所有方法都标了事务注解，事务管理器名默认 springCloudTaskTransactionManager：
/**
* Notifies the repository that a taskExecution needs to be created.
* @param taskExecution a TaskExecution instance containing the startTime, arguments
* and externalExecutionId that will be stored in the repository. Only the values
* enumerated above will be stored for this TaskExecution.
* @return the {@link TaskExecution} that was stored in the repository. The
* TaskExecution's taskExecutionId will also contain the id that was used to store the
* TaskExecution.
*/
@Transactional("${spring.cloud.task.transaction-manager:springCloudTaskTransactionManager}")
TaskExecution createTaskExecution(TaskExecution taskExecution);
/**
* Creates an empty TaskExecution with just an id and name provided. This is intended
* to be utilized in systems where the request of launching a task is separate from
* the actual start of a task (the underlying system may need to deploy the task prior
* to launching, etc).
* @param name task name to be associated with the task execution.
* @return the initial {@link TaskExecution}
*/
@Transactional("${spring.cloud.task.transaction-manager:springCloudTaskTransactionManager}")
TaskExecution createTaskExecution(String name);
/**
* Creates an empty TaskExecution with just an id provided. This is intended to be
* utilized in systems where the request of launching a task is separate from the
* actual start of a task (the underlying system may need to deploy the task prior to
* launching, etc).
* @return the initial {@link TaskExecution}
*/
@Transactional("${spring.cloud.task.transaction-manager:springCloudTaskTransactionManager}")
TaskExecution createTaskExecution();
/**
* Notifies the repository that a taskExecution has has started.
* @param executionid to the task execution to be updated.
* @param taskName the name that associated with the task execution.
* @param startTime the time task began.
* @param arguments list of key/value pairs that configure the task.
* @param externalExecutionId id assigned to the task by the platform.
* @return TaskExecution created based on the parameters.
*/
@Transactional("${spring.cloud.task.transaction-manager:springCloudTaskTransactionManager}")
TaskExecution startTaskExecution(long executionid, String taskName, LocalDateTime startTime, List<String> arguments,
String externalExecutionId);
/**
* Notifies the repository to update the taskExecution's externalExecutionId.
* @param executionid to the task execution to be updated.
* @param externalExecutionId id assigned to the task by the platform.
*/
@Transactional("${spring.cloud.task.transaction-manager:springCloudTaskTransactionManager}")
void updateExternalExecutionId(long executionid, String externalExecutionId);schema 由 schema-h2.sql（以及 mysql/oracle/postgres 等变体）定义：TASK_EXECUTION、TASK_EXECUTION_PARAMS（参数键值）、TASK_TASK_BATCH（与 Batch 执行关联）、序列 TASK_SEQ，以及单实例锁表 TASK_LOCK（LOCK_KEY+REGION 主键、CLIENT_ID、CREATED_DATE、EXPIRED_AFTER）：
CREATE TABLE TASK_EXECUTION (
TASK_EXECUTION_ID BIGINT NOT NULL PRIMARY KEY ,
START_TIME TIMESTAMP(9) DEFAULT NULL ,
END_TIME TIMESTAMP(9) DEFAULT NULL ,
TASK_NAME VARCHAR(100) ,
EXIT_CODE INTEGER ,
EXIT_MESSAGE VARCHAR(2500) ,
ERROR_MESSAGE VARCHAR(2500) ,
LAST_UPDATED TIMESTAMP(9),
EXTERNAL_EXECUTION_ID VARCHAR(255),
PARENT_EXECUTION_ID BIGINT
);
CREATE TABLE TASK_EXECUTION_PARAMS (
TASK_EXECUTION_ID BIGINT NOT NULL ,
TASK_PARAM VARCHAR(2500) ,
constraint TASK_EXEC_PARAMS_FK foreign key (TASK_EXECUTION_ID)
references TASK_EXECUTION(TASK_EXECUTION_ID)
) ;
CREATE TABLE TASK_TASK_BATCH (
TASK_EXECUTION_ID BIGINT NOT NULL ,
JOB_EXECUTION_ID BIGINT NOT NULL ,
constraint TASK_EXEC_BATCH_FK foreign key (TASK_EXECUTION_ID)
references TASK_EXECUTION(TASK_EXECUTION_ID)
) ;
CREATE SEQUENCE TASK_SEQ ;
CREATE TABLE TASK_LOCK (
LOCK_KEY CHAR(36) NOT NULL,
REGION VARCHAR(100) NOT NULL,
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP(9) NOT NULL,
EXPIRED_AFTER TIMESTAMP(9) NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);表初始化由 spring.cloud.task.initialize-enabled 控制（TaskProperties.java:92-97，null 表示按嵌入式数据库自动判断）；配错会导致表缺失、启动失败——这是 Task 最常见的“环境问题”。initialize-enabled 只建表，不迁移既有表结构。
单实例锁不是调度协调：启用 single-instance-enabled=true 后，SingleInstanceTaskListener 在启动回调通过 JdbcLockRegistry 按 task name 竞争；共享同库、锁前缀和任务名的不同执行实例才形成互斥。竞争失败事件使第二次启动抛 TaskExecutionException，不能当作 FIFO 排队直至前任务完成；结束/失败回调释放锁。尤其要注意固定5.0.2的单位冲突：TaskProperties 注释称 single-instance-lock-ttl 为毫秒，但 listener 实际调用 Duration.ofSeconds(value)，Integration7的该参数就是过期时间。默认 Integer.MAX_VALUE 因此约68年，而非真正无限；本实验显式300秒、轮询50毫秒。生产恢复窗口需按实际实现验证，必要时提供明确 Duration 的 LockRegistry，不能照错误注释调小数值。它不 fence 业务副作用，也不提供调度、故障转移或 exactly-once；本票验证一个 JVM 内两个线程/独立 Context 的共享 H2 文件库竞争，未验证跨进程和 TTL 到期恢复。
同一固定版本的属性注释与实际Duration构造不一致，必须一起读：
*/
private boolean singleInstanceEnabled = false;
/**
* Declares the maximum amount of time (in millis) that a task execution can hold a
* lock to prevent another task from executing with a specific task name when the
* single-instance-enabled is set to true. Default time is: Integer.MAX_VALUE.
*/
private int singleInstanceLockTtl = Integer.MAX_VALUE;
/**
* Declares the time (in millis) that a task execution will wait between checks.
* Default time is: 500 millis.
*/
private int singleInstanceLockCheckInterval = DEFAULT_CHECK_INTERVAL; }
}Task 的完整启动与失败输入
Task 使用独立工程，不把长运行的 Stream 应用直接当短任务。沿用前文 POM 的 parent/BOM/插件，artifactId 改为 task-demo，dependencies 改为下面四项（其他 Stream/Function 依赖移除）：
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-task</artifactId></dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jdbc</artifactId></dependency>
<dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-jdbc</artifactId></dependency>
<dependency><groupId>com.h2database</groupId><artifactId>h2</artifactId><scope>runtime</scope></dependency>// src/main/java/study/task/TaskDemo.java；教学代码，失败输入为 --lab.fail=true。
package study.task;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.task.configuration.EnableTask;
import org.springframework.cloud.task.listener.TaskExecutionListener;
import org.springframework.cloud.task.repository.TaskExecution;
import org.springframework.context.annotation.Bean;
import org.springframework.core.env.Environment;
@EnableTask
@SpringBootApplication
public class TaskDemo {
public static void main(String[] args) { SpringApplication.run(TaskDemo.class, args); }
@Bean
public CommandLineRunner work(Environment environment) {
return args -> {
if (environment.getProperty("lab.fail", Boolean.class, false)) {
throw new IllegalStateException("synthetic task failure");
}
System.out.println("synthetic work finished");
};
}
@Bean
public TaskExecutionListener lifecycle() {
return new TaskExecutionListener() {
@Override
public void onTaskStartup(TaskExecution execution) {
System.out.println("task execution id=" + execution.getExecutionId());
}
@Override
public void onTaskFailed(TaskExecution execution, Throwable failure) {
execution.setExitMessage("synthetic failure; inspect task error record");
}
};
}
}# src/main/resources/application.yml；LAB_TASK_DB 是本次独有目录下的文件基名。
spring:
application:
name: u36-task-demo
main:
web-application-type: none
datasource:
url: jdbc:h2:file:${LAB_TASK_DB}
username: sa
password: ""
cloud:
task:
initialize-enabled: true
closecontext-enabled: true
single-instance-enabled: true新建一个独有临时目录，把其下的 task 文件基名传给 LAB_TASK_DB，先运行 mvn spring-boot:run，再用同一基名运行 mvn spring-boot:run -Dspring-boot.run.arguments=--lab.fail=true。预期出现两条不同 execution id：正常行有开始/结束时间及退出码 0，异常行有错误信息与非零退出码；第二次启动是一次新的执行，不从第一行的业务中断处自动续跑。使用公开 TaskExplorer 查询这两条记录，比控制台中看到一行“finished”更能说明持久簿记。closecontext-enabled 默认 false，本例显式 true，使短任务结束时关闭上下文和连接池；initialize-enabled=true 只适用于这个隔离新库，既有业务数据库应由受控迁移预建表，不靠它升级 schema。
单实例验证必须确保两个启动方看到同一个数据库、表前缀和 task name；每次改成新的 H2 内存库，就根本不存在互斥的共同记录。上面的普通 H2 文件 URL 适合顺序运行，不能据此声称两个独立 JVM 已并发共享数据库；跨进程验证须另配共享数据库连接或专用 H2 server，并单列运行结果。锁 TTL 也不是一次任务的截止时间，更不能替代业务去重：锁释放后相同输入仍可再次执行。若进程被强制终止，恢复方应先核对外部进程与业务状态，再决定新建执行/人工补偿，不能只看到未结束的行就并发重启。
Task 启动边界
三个容易混淆的“任务”概念：
| 概念 | 谁触发 | 有没有执行记录/退出码 | 归属 |
|---|---|---|---|
@Scheduled | 进程内调度器按 fixedDelay/fixedRate/cron | 无持久记录、无退出码、重启不补跑 | 调度执行：@Scheduled 语义、异常吞噬、关闭与观测（U21/U23） |
| Spring Cloud Task | 外部（CI/K8s Job/调度器/消息）启动进程 | 有 TASK_EXECUTION 行、退出码、监听器、可选单实例锁 | 本页 |
| Spring Batch / Quartz | JobLauncher / 调度器 | 各有自己的元数据表（BATCH_JOB_EXECUTION 等） | U46 |
@Scheduled 的方法可以跑在一个 @EnableTask 应用里，但两者没有机制关系：调度器不写 TASK_EXECUTION，Task 不感知调度周期。“经消息启动 Task” 在本轮固定的 Task 5.0.2 中并不存在一个现成的 TaskLauncher 类：spring-cloud-task-stream 模块只包含 TaskEventAutoConfiguration——它用 StreamBridge 把 TaskExecution 的生命周期事件（startup/end/failed）发到绑定 task-events（TaskEventAutoConfiguration.java:51-82，TaskEventProperties.java:84）。也就是说 Task 与 Stream 的集成方向是Task 发事件给 Stream，不是 Stream 收消息拉起 Task；后者需要外部启动器（例如 Data Flow 的部署器）自己实现，且重复的启动消息会带来重复执行，需要业务幂等（U61）。经 Stream 启动 Task 的运行行为本页未运行/推演。
Task与Batch/Stream桥接：一条运行记录不等于一个Job或一条消息
Task的记录单位是一次短生命周期应用执行，Batch的记录单位包含JobInstance、JobExecution、StepExecution与checkpoint。一个Task里可跑多个Job；一次失败Job的重启可能产生新JobExecution与新TaskExecution，两套ID不应混作业务幂等键。spring-cloud-task-batch 的桥接listener在Task启动时记住TaskExecution，在beforeJob里把关联存入TaskBatchDao，提供可查询的运行关系，没有把两套状态机合并：
TaskBatchExecutionListener.java:52–66（固定commit，SHA-256见元数据）：
@Override
public void onTaskStartup(TaskExecution taskExecution) {
this.taskExecution = taskExecution;
}
@Override
public void beforeJob(JobExecution jobExecution) {
if (this.taskExecution == null) {
logger.warn("This job was executed outside the scope of a task but still used the task listener.");
}
else {
logger.info(String.format("The job execution id %s was run within the task execution %s",
jobExecution.getId(), this.taskExecution.getExecutionId()));
this.taskBatchDao.saveRelationship(this.taskExecution, jobExecution);
}Task Batch与single-step starter的完整checkpoint/chunk/重启机制归U46，本节必须掌握它们的接点。下面是未运行的固定API教学工程：独立Boot4.0.9/Cloud2025.1.4项目加入：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-single-step-batch-job</artifactId>
<version>5.0.2</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-task-batch</artifactId>
<version>5.0.2</version>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<scope>runtime</scope>
</dependency>single-step starter带入Task core、Boot Batch/JDBC等基础依赖，本例另显式加入Task Batch以记录两套ID关系；普通Batch工程可单独选择 spring-cloud-task-batch:5.0.2 接关系，不需为了关联ID强行改成单步。完整入口 src/main/java/study/batchbridge/BridgeApp.java：
package study.batchbridge;
import java.util.List;
import java.util.Map;
import org.springframework.batch.infrastructure.item.ItemReader;
import org.springframework.batch.infrastructure.item.ItemWriter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.task.configuration.EnableTask;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
@EnableTask
public class BridgeApp {
public static void main(String[] args) {
SpringApplication.run(BridgeApp.class, args);
}
@Bean
ItemReader<Map<String, Object>> itemReader() {
var rows = List.<Map<String, Object>>of(
Map.of("id", 1L, "value", "alpha"),
Map.of("id", 2L, "value", "beta")).iterator();
return () -> rows.hasNext() ? rows.next() : null;
}
@Bean
ItemWriter<Map<String, Object>> itemWriter() {
return chunk -> {
for (Map<String, Object> row : chunk) {
System.out.println(row);
}
};
}
}注意Batch6的ItemReader/Writer是 org.springframework.batch.infrastructure.item，不能照搬Batch5旧包名。配置：
spring.application.name=u36-batch-bridge
spring.main.web-application-type=none
spring.datasource.url=jdbc:h2:mem:u36bridge;DB_CLOSE_DELAY=-1
spring.datasource.username=sa
spring.datasource.password=
spring.cloud.task.initialize-enabled=true
spring.batch.jdbc.initialize-schema=always
spring.batch.job.job-name=u36Job
spring.batch.job.step-name=u36Step
spring.batch.job.chunk-size=2
spring.cloud.task.batch.fail-on-job-failure=true预期自动生成一个单步Job，reader返回null结束，写两个Map，关联TaskExecution与JobExecution，正常结束Task exitCode=0。内存reader没有ItemStream checkpoint，内存H2重启丢元数据，println也不能回滚；这组样例只演示桥接，不能当重启或数据库业务恰好一次证明。固定自动配置先验证jobName/stepName非空和chunkSize>0，再用JobRepository/transactionManager、reader、可选processor、writer构造一个chunk step：
SingleStepJobAutoConfiguration.java:70–91（固定commit，SHA-256见元数据）：
private void validateProperties(SingleStepJobProperties properties) {
Assert.hasText(properties.getJobName(), "A job name is required");
Assert.hasText(properties.getStepName(), "A step name is required");
Assert.notNull(properties.getChunkSize(), "A chunk size is required");
Assert.isTrue(properties.getChunkSize() > 0, "A chunk size greater than zero is required");
}
@Bean
@ConditionalOnMissingBean
@ConditionalOnProperty(prefix = "spring.batch.job", name = "job-name")
public Job job(ItemReader<Map<String, Object>> itemReader, ItemWriter<Map<String, Object>> itemWriter) {
SimpleStepBuilder<Map<String, Object>, Map<String, Object>> stepBuilder = new StepBuilder(
this.properties.getStepName(), this.jobRepository)
.<Map<String, Object>, Map<String, Object>>chunk(this.properties.getChunkSize(), this.transactionManager)
.reader(itemReader);
stepBuilder.processor(this.itemProcessor);
Step step = stepBuilder.writer(itemWriter).build();
return new JobBuilder(this.properties.getJobName(), this.jobRepository).start(step).build();BatchStatus.FAILED、Batch ExitStatus、Task exitCode与OS退出码是不同观测点。默认Boot Batch runner可能捕获Job执行失败并把FAILED保存下来，而应用正常退出，Task会记录0。spring.cloud.task.batch.fail-on-job-failure=true 选择Task的runner，把观察到的FAILED转换成TaskException；异步Job还需等待并轮询，默认间隔5000ms，由 spring.cloud.task.batch.fail-on-job-failure-poll-interval 调整。它并不提供超时取消、自动重试或调度器。OS非0仍要看入口是否传播或采用exit策略。
最小故障矩阵（均为未运行预期）：chunk-size=0应在配置验证时报错；writer遇id=2抛错应产生FAILED与对应Task失败；关掉fail-on-job-failure后同时核对Batch FAILED与Task退出码，不能只看进程结束日志；相同Job名和identifying parameters完成后再启动，应按Batch的JobInstance完成约束处理，而非删除元数据“解决”。恢复必须根据Task→Job关联找到真实失败Step与参数，确认业务副作用是否提交，再用U46的restart策略；新Task ID本身不允许重复扣款。
spring-cloud-task-stream:5.0.2 则把Task/Batch监听回调发送到StreamBridge。独立普通Task工程使用 spring-cloud-starter-task、spring-cloud-task-stream 各5.0.2，另选 spring-cloud-stream-binder-kafka:5.0.4 和运行时H2；应用仍需 @EnableTask，使用本页Task的CommandLineRunner入口；追加下列完整目的地配置，不与上面的single-step实验混成一个未经核验的启动形态：
spring.cloud.task.events.enabled=true
spring.cloud.task.batch.events.task-event-binding-name=task-events
spring.cloud.stream.output-bindings=task-events
spring.cloud.stream.bindings.task-events.destination=u36-task-events
spring.cloud.stream.bindings.task-events.content-type=application/json
spring.cloud.stream.kafka.binder.brokers=${KAFKA_BOOTSTRAP:127.0.0.1:9092}开关前缀是 spring.cloud.task.events，但task-event-binding-name所在属性类前缀实际为 spring.cloud.task.batch.events，不要按直觉发明配置。默认binding名为task-events。固定监听器在startup/end/failed直接 streamBridge.send，没有Outbox或“发送与Task元数据库完成原子提交”：
TaskEventAutoConfiguration.java:64–75（固定commit，SHA-256见元数据）：
public void onTaskStartup(TaskExecution taskExecution) {
streamBridge.send(taskEventProperties.getTaskEventBindingName(), taskExecution);
}
@Override
public void onTaskEnd(TaskExecution taskExecution) {
streamBridge.send(taskEventProperties.getTaskEventBindingName(), taskExecution);
}
@Override
public void onTaskFailed(TaskExecution taskExecution, Throwable throwable) {
streamBridge.send(taskEventProperties.getTaskEventBindingName(), taskExecution);Task的end/failed回调在TaskRepository最终complete之前运行，所以end事件不是“查询端此刻必已读到DB endTime”的保证。原始/失败/结束消息中只看到TaskExecution，不应未经协议设计就假定含统一eventType；失败会触发failed再end，消费端须按executionId与状态/业务协议区分及幂等。发出end后DB完成写失败，或DB有记录但消息异步失败，都会让两边状态不一致。业务完成事件要求U61的Outbox或明确补偿/对账，不能靠增加重试次数证明原子性。
Batch事件自动配置另要求存在Job与TaskLifecycleListener、spring.cloud.task.batch.events.enabled=true（默认true），有job/step/chunk/item-read/item-process/item-write/skip等监听输出。该固定版还以原始占位符 ${spring.batch.job.jobName:} 为空作条件，single-step starter设置的是宽松绑定的job-name；命名形态、配置源、条件求值与监听器存在性必须实测，不能仅说“加stream依赖每种Batch任务必然发全部事件”。本节未运行Task→Stream或Batch事件；实际验收需同时记录Task/Job ID、binding启动条件、topic里的原始消息和DB最终行，不以StreamBridge返回true替代broker消费证据。
容量与选择（假设）：200个Task/s，每个成功Task至少startup/end两条通知，基础事件量400条/s；失败时再多failed，按每item发送事件可能放大到远超Task数。只查任务历史可直接用TaskExplorer/Batch JobExplorer，少一条broker故障链；有跨进程监控消费需求再选Task Stream。业务编排、调度平台与分布式业务事务并不是这些listener的职责。固定来源：Task Batch桥接、single-step starter、Task Stream。
Contract：DSL 与 verifier
再次消歧：本节的 Contract 是 Spring Cloud Contract 5.0.3（消费者驱动契约测试），与 Feign 的 Contract SPI 无关。维护状态见首节：EOL、spring-attic 归档、5.0.3 为最后 patch。以下内容是“固定版本机制 + 迁移案例”，不是推荐在维护项目中使用。
契约 DSL 的入口是 Contract.make(Consumer<Contract>)（Java DSL）或 Groovy closure 版本；契约定义 request（method/url/body/headers/matchers）与 response（status/body/matchers），消息契约则用 label/inputMessage/outputMessage：
/**
* Point of entry to build a contract.
* @param consumer function to manipulate the contract
* @return manipulated contract
*/
public static Contract make(Consumer<Contract> consumer) {
Contract contract = new Contract();
consumer.accept(contract);
return contract;
}
/**
* Groovy point of entry to build a contract. Left for backward compatibility reasons.
* @param closure function to manipulate the contract
* @return manipulated contract
*/
public static Contract make(@DelegatesTo(Contract.class) Closure closure) {
Contract dsl = new Contract();
closure.setDelegate(dsl);
closure.call();
Contract.assertContract(dsl);
return dsl;
}verifier 的 Maven/Gradle 插件扫描 src/test/resources/contracts，为每份契约在 target/generated-test-sources/contracts 生成 JUnit 测试（基于 baseClass 继承生产者测试基类）。TestGenerator.generateTestClasses 先递归扫描契约文件、检查 in-progress 标记（failOnInProgress=true 时发现 in-progress 契约直接失败），再逐个生成：
void generateTestClasses(final String basePackageName) {
MultiValueMap<Path, ContractMetadata> contracts = contractFileScanner.findContractsRecursively();
log.debug("Found the following contracts {}", contracts.keySet());
Set<Map.Entry<Path, List<ContractMetadata>>> inProgress = inProgress(contracts);
if (!inProgress.isEmpty() && configProperties.isFailOnInProgress()) {
String inProgressContractsPaths = inProgress.stream()
.map(Map.Entry::getKey)
.map(Path::toString)
.collect(Collectors.joining(","));
throw new IllegalStateException("In progress contracts found in paths [" + inProgressContractsPaths
+ "] and the switch [failOnInProgress] is set to [true]. Either unmark those contracts as in progress, or set the switch to [false].");
}
processAll(contracts, basePackageName);
}
private Set<Map.Entry<Path, List<ContractMetadata>>> inProgress(MultiValueMap<Path, ContractMetadata> contracts) {
return contracts.entrySet()
.stream()
.filter(entry -> entry.getValue().stream().anyMatch(ContractMetadata::anyInProgress))
.collect(Collectors.toSet());
}生成的测试把契约里的 request 发到生产者（本实验为 MockMvc），断言 response 满足契约。公开入口实际生成 validate_order_ready：READY 响应通过，改为 BROKEN 后生成的 JSON 断言失败，恢复后再通过。10条通过用例与1条预期失败负控的报告分别保存，负控不是依赖下载错误。它只证明被执行的生产者行为符合这份契约，不证明消费者实际按契约调用或生产部署兼容。Contract5.0.3已明确钉版本且为EOL历史教学基线，不能由本组合成功推出当前官方维护。
Contract 的生成与消费示例
以下是固定 5.0.3 的历史机制示例，不把 EOL 项目当作新系统推荐。生产者使用独立 Maven 工程，沿用首节的 Boot parent，坐标改为 study:greeting-producer:1.0.0；dependencies 使用 Boot 管理的 spring-boot-starter-webmvc、测试 scope 的 spring-boot-starter-test，并显式加入：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-contract-verifier</artifactId>
<version>5.0.3</version><scope>test</scope>
</dependency>在 build/plugins 增加契约插件，插件版本与测试库分别固定，不能指望 Cloud 2025.1.4 BOM 为它们补版本：
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-contract-maven-plugin</artifactId>
<version>5.0.3</version><extensions>true</extensions>
<configuration>
<testFramework>JUNIT5</testFramework>
<testMode>MOCKMVC</testMode>
<baseClassForTests>study.contract.ContractBase</baseClassForTests>
</configuration>
</plugin>文件 src/test/resources/contracts/greeting.groovy 描述消费者依赖的响应，而不是复制生产者实现：
org.springframework.cloud.contract.spec.Contract.make {
name 'greeting'
request {
method GET()
url '/greeting'
}
response {
status OK()
headers { contentType(applicationJson()) }
body([message: 'hello'])
}
}生产者代码与生成测试基类如下，分别放在 src/main/java/study/contract/GreetingController.java、src/test/java/study/contract/ContractBase.java。基类接入真实 controller 的 MockMvc 边界，没有在测试里另写一个返回相同常量的假 controller：
package study.contract;
import java.util.Map;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class GreetingController {
@GetMapping("/greeting")
public Map<String, String> greeting() { return Map.of("message", "hello"); }
}package study.contract;
import io.restassured.module.mockmvc.RestAssuredMockMvc;
import org.junit.jupiter.api.BeforeEach;
public abstract class ContractBase {
@BeforeEach
void configureProducer() {
RestAssuredMockMvc.standaloneSetup(new GreetingController());
}
}mvn clean verify 的预期是生成测试并验证 controller，生成位置为 target/generated-test-sources/contracts；mvn install 还把附加的 stubs 制品安装到本地 Maven 仓库。故意把 controller 的响应键改成 text，契约仍要求 message，应使生成测试失败；修复回 controller 后重新生成并执行，才是一次闭环。改契约去迎合所有实现变化会失去漂移检测价值，契约变更须由生产者与消费者共同评审。
消费者独立测试工程加入显式版本的 spring-cloud-starter-contract-stub-runner:5.0.3（scope=test）和 Boot 的 spring-boot-starter-test。以下 JUnit 测试使用前一步安装的 study:greeting-producer:1.0.0:stubs，不下载 REMOTE 制品，不占用固定端口：
package study.consumer;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.contract.stubrunner.spring.AutoConfigureStubRunner;
import org.springframework.cloud.contract.stubrunner.spring.StubRunnerPort;
import org.springframework.cloud.contract.stubrunner.spring.StubRunnerProperties;
import static org.junit.jupiter.api.Assertions.assertEquals;
@SpringBootTest(classes = StubConsumerTest.Config.class,
webEnvironment = SpringBootTest.WebEnvironment.NONE)
@AutoConfigureStubRunner(ids = "study:greeting-producer:1.0.0:stubs",
stubsMode = StubRunnerProperties.StubsMode.LOCAL)
class StubConsumerTest {
@SpringBootConfiguration @EnableAutoConfiguration
static class Config { }
@StubRunnerPort("greeting-producer") int port;
@Test
void readsGreeting() throws Exception {
var request = HttpRequest.newBuilder(
URI.create("http://127.0.0.1:" + port + "/greeting"))
.timeout(Duration.ofSeconds(2)).GET().build();
try (var client = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(2)).build()) {
var response = client.send(request, HttpResponse.BodyHandlers.ofString());
assertEquals(200, response.statusCode());
assertEquals("{\"message\":\"hello\"}", response.body());
}
}
}这里验证的是固定 stub 的 HTTP 边界。缺少 stubs 制品时先核对 Maven 坐标、classifier、版本与本地仓库，不能把下载失败解释为生产者违反契约；stub 测试通过后，生产者自身的生成测试仍必须运行，才能让两端指向同一份契约。真实网络鉴权、部署差异、容量、broker 确认与跨资源事务均需要各自的集成验证。
StubRunner 与消息契约边界
StubRunner 是契约测试的另一半：把生产者发布的 stub jar（-stubs.jar）在消费者测试中启动为真实 HTTP stub（WireMock 3.13.2，随机 loopback 端口），消费者用自己的真实客户端调用。StubRunner.runStubs 调本地 runner 解析 stub 仓库与配置，返回 RunningStubs（含每个 stub 的端口），可选把注册的 mappings 导出到目录：
@Override
public RunningStubs runStubs() {
registerShutdownHook();
RunningStubs stubs = this.localStubRunner.runStubs(this.stubRunnerOptions, this.stubRepository,
this.stubsConfiguration);
if (this.stubRunnerOptions.hasMappingsOutputFolder()) {
String registeredMappings = this.localStubRunner.registeredMappings();
if (StringUtils.hasText(registeredMappings)) {
File outputMappings = new File(this.stubRunnerOptions.getMappingsOutputFolder(),
this.stubsConfiguration.artifactId + "_"
+ stubs.getPort(this.stubsConfiguration.toColonSeparatedDependencyNotation()));
try {
outputMappings.getParentFile().mkdirs();
clearOldFiles(outputMappings.getParentFile(), this.stubsConfiguration.artifactId);
outputMappings.createNewFile();
Files.write(Paths.get(outputMappings.toURI()), registeredMappings.getBytes());
if (log.isDebugEnabled()) {
log.debug("Stored the mappings for artifactid [" + this.stubsConfiguration.artifactId + "] at ["
+ outputMappings + "] location");
}
}
catch (IOException e) {
log.error("Exception occurred while trying to store the mappings", e);
throw new IllegalStateException(e);
}
}
}
return stubs;
}@AutoConfigureStubRunner 将 stub 生命周期接到 Spring 测试上下文。stubsMode=CLASSPATH（默认）从 classpath 取 stub，LOCAL 从本地 Maven 仓库取，REMOTE 从远程位置解析；这些不是同一个模式。公开实验以插件从同一DSL生成的 mapping 启动真实 loopback HTTP stub，并验证200与未知路径404；没有验证远程制品仓库解析。stub 是测试替身，消费者调用通过不证明真实服务的网络、授权与副作用；定位失败先区分制品/端口/路径配置和行为契约。
消息契约
消息契约用 label、触发输入与 outputMessage 的 destination/payload 描述消息行为，由 ContractVerifierMessaging 对接具体消息实现。它和HTTP契约的证明对象不同，不能笼统比较“语义更弱”。test binder仅支持被执行的函数、载荷与目的地匹配；真实broker实验才能支持相应序列化与交付路径，二者均不自动证明复制、幂等或生产可靠性。本轮消息契约与 REMOTE stub 模式均未运行。
消息契约的配置面可用一个发出 greeting 事件的例子理解，以下仍为推演，不是已经可运行的 Kafka 契约实验：
// src/test/resources/contracts/greetingEvent.groovy
org.springframework.cloud.contract.spec.Contract.make {
label 'greeting_event'
input { triggeredBy('publishGreeting()') }
outputMessage {
sentTo('greetings')
headers { messagingContentType(applicationJson()) }
body([eventId: 'synthetic-u36-1', message: 'hello'])
}
}这里 publishGreeting() 是生成测试基类必须实现的触发方法，其方法体调用生产者的真实发送入口；sentTo('greetings') 则是所选 verifier adapter 识别的目的地。若接 Stream，应用的 spring.cloud.stream.bindings.<output>.destination 必须与 adapter 对目的地的解释一致；若接 Kafka 直连，则还需 spring.kafka.bootstrap-servers、serializer/deserializer、group 与 topic 的专用配置。基类按所选消息实现启用 @AutoConfigureMessageVerifier，并提供能向该实现发送/接收的 verifier，不能仅添加 DSL 就假定存在 Kafka 消费器。超时必须有界，消息的 eventId、body、content type 要一起核对；触发有输出而 verifier 等不到时，先检查 binding 名与物理 destination、序列化和实际 adapter，而不是立刻宣布“丢消息”。
固定 5.0.3 的上游 Kafka Stub Runner 样例本身有迁移相关禁用标记，本页因此不把该旧样例当作此组合可运行的证据。把消息契约接成真实 broker 验证，需要单独实测 adapter、依赖组合、启动与清理；本票的 HTTP verifier/stub 结果不能替它背书。
三个stub解析模式在固定源码中分别列为独立枚举：
public enum StubsMode {
/**
* Pick the stubs from classpath.
*/
CLASSPATH,
/**
* Fetch the stubs from local .m2.
*/
LOCAL,
/**
* Fetch the stubs from a remote location.
*/
REMOTE,框架词义 vs broker 实际保证
全页的收口表。每一行的“保证方”是不同层，评审时先问“这句话由谁保证”，再问“证据在哪一层”。
| 词 | Stream/Function 层说法 | 实际保证方 | 不能推出 |
|---|---|---|---|
StreamBridge.send 返回 true | 默认同步channel分派且Kafka sync=false时仅接受；同步分派+sync=true时另含发送Future完成；StreamBridge async另论 | channel、发送handler及acks配置 | 无条件推出broker复制、业务处理或跨资源提交 |
ackMode 提交 | 容器在策略时机提交已交付前缀 | Spring Kafka 容器 → Kafka 消费者提交协议（U57） | 业务效果已落库；MANUAL 下应用确认即责任 |
| committed offset | 恢复起点 | Kafka __consumer_offsets（U57） | 记录已成功处理 |
enableDlq | 失败记录 republish 到 error.<dest>.<group> | Spring Kafka客户端；是否与offset原子取决于Kafka事务路径 | Kafka内建死信协议、外部数据库原子提交或业务恢复完成 |
maxAttempts/backOff* | binder 内 RetryTemplate 重试 | Spring Retry + 消费循环 | broker 延迟重投；重试不占 poll 时间 |
transaction-id-prefix | 消费-生产链路的 Kafka 事务 | Spring Kafka KafkaTransactionManager → broker 事务（U57） | 跨数据库 EOS；业务幂等 |
| Task 退出码 | 本次Task执行记录的结束状态 | Task处理ExitCodeEvent/监听异常并写入TASK_EXECUTION | 自动等同OS退出码、业务成功、崩溃时必有end行 |
| Contract 生成测试通过 | 生产者满足契约 | 生成的 JUnit 测试 + stub/生产者 | 生产环境兼容；消费者真实调用正确 |
这张表与 MQ生产存储消费可靠性 的 七个成功点，七种承诺 是同一认识论：每加一层框架，词义就多一层，但保证不会自动叠加。方法论文档（K04）要求“区分规范、实现、配置、版本和故障模型”，本页的做法是每个结论都标层：源码（固定提交）、官方文档（同版本）、实验（以证据为准）或推演。
容量、重试预算与恢复责任
下面是带假设的纸面算例，所有时延均为输入条件，不是本机实测。一个命令式事件处理服务每秒收到 600 条消息，12 个分区，部署总共 6 个消费者线程，每线程顺序处理，成功处理时间为 5ms/条；忽略序列化与提交开销时，正常处理上限为 6 / 0.005 = 1200条/秒，正常负载占该假设上限 50%。这只是平均服务时间与并发数推导，不能推出 P99、broker 吞吐或生产容量。
现在假设 1% 输入会持续触发下游 80ms 超时，仍使用 binder max-attempts=3、退避 1s/2s。每条失败记录占用消费者线程 3 × 0.080 + 1 + 2 = 3.24秒。平均占用变成 0.99 × 0.005 + 0.01 × 3.24 = 0.03735秒/条，六线程的处理能力约 6 / 0.03735 ≈ 161条/秒，小于到达率 600条/秒；积压约增加 439条/秒。即使 broker 还很空闲，“只失败 1%”也足以耗尽消费线程。如果一次 poll 的 100 条都落入这个失败分支，处理需约 324s，超过示例显式设置的 max.poll.interval.ms=300000（300s）；重复投递和重平衡风险来自这个时间预算，而不是把 maxAttempts 改小就自动消失。
| 方案 | 适用条件与代价 | 必须承担的恢复工作 |
|---|---|---|
| A：消费路径内短重试 | 能证明是短暂故障、处理幂等、重试和处理总时间可放进 poll/请求预算；少建重试基础设施，但阻塞同线程后续记录，甚至同批其他分区 | 应用设置总尝试/退避预算、区分不可重试错误；运维观察处理时间/重平衡/lag，超预算后转 DLQ 或显式恢复路径 |
| B：把失败记录交给有界 retry topic/持久调度，再推进主路径 | 可接受异步恢复和暂时乱序，需重试队列、调度、去重和最终失败处置；不能把同分区业务强顺序作为无条件保证 | 发送方确认持久交接结果；应用维护 eventId/尝试次数/下次时间和幂等；恢复方限速回放，核对 DLQ 与业务终态，避免恢复风暴 |
方案 B 的好处也必须在条件下计算：若主路径失败仅做一次 80ms 尝试，持久交接另耗 5ms，则平均占用约 0.99 × 0.005 + 0.01 × 0.085 = 0.0058秒/条，主路径理论上限约 1034条/秒。但这假设交接已得到所需确认，且没有把重试工作凭空消除；重试系统仍须消化每秒 6 条新失败及历史积压。若故障恢复后要在 10 分钟内清掉 18000 条积压，最低净清理速率为 18000 / 600 = 30条/秒，还要加每秒新增失败和下游允许的负载，不能无界回放。Kafka 原子消费-生产可以改变交接失败窗口，跨数据库的业务效果仍需 U61 的持久意图与幂等。
演进触发可写成可检查的条件：处理时间/退避接近 poll 预算、单线程 head-of-line blocking 影响无关记录、DLQ 长期增长且回放责任不明，便重新评估重试位置；不能只因“用了 Stream”就默认选择 A 或 B。验证先在合成小流量下固定输入比例、时间窗口、分区数与线程数，记录成功/重试/最终失败计数与 lag，再注入持续故障并恢复；每一次重放都核对业务不变量，不能仅凭消费数归零认定恢复成功。具体 broker 模型归 U57，Outbox/Inbox 的事务边界归 U61。
Task 与 Contract 的选型也有成本边界。假设每天生成 10000 份合成报表、每份 20s，允许失败后在 1h 内补齐，业务不变量是同一 reportDate/customerId 只发布一个最终版本。Task+外部调度器适合“一次应用启动就是一份或一批报表”，每次启动有记录、退出码和运行资源成本，重试和未结束执行的对账归外部控制方；Batch 的 chunk/step 重启模型适合一个大批次中有许多可断点步骤，需要额外的 job/step 元数据与事务设计，主讲在 U46。仅凭 Task 退出码 0，既不能证明报表已发布，也不能保证强制终止后已自动恢复；业务唯一键、产物校验与发布状态必须另查。
这个负载每天至少需要 10000 × 20 = 200000执行秒，均匀到达且忽略启动成本时平均在途约 200000 / 86400 ≈ 2.32份。如果“每份一个 Task”却全部使用相同 task name 开单实例锁，串行上限只有 86400 / 20 = 4320份/天，连日常输入都处理不完。可按有业务含义的分片建立多个执行槽，或让一次 Task 内由有界执行器处理一批，代价分别是分片调度与进程内取消/结果聚合；两者都需保留业务唯一发布约束。如果一次故障积压了 10000 份并要求 1h 内全部补完，仅计算报表执行就至少需要 ceil(10000 × 20 / 3600) = 56 个持续工作槽，还没有计入正常新任务、启动/失败开销与下游配额；恢复方必须先确认这个期限与资源是否同时可满足。
若报表依赖外部服务契约，固定 Contract 5.0.3 能保存旧消费者约定并在已有项目中检测漂移，代价是 EOL 工具的运行时与安全维护责任。另一合理选择是保留同一份行为样例与失败案例，迁移到受维护的契约/集成验证工具；迁移必须双跑已知正确响应与故意删字段的负例，验证新工具仍能捕获原有破坏性变更，然后按双方确认的版本退出旧工具。不能为了升级工具把有业务意义的旧契约直接删掉，也不能因 stub 测试通过就省掉真实生产者和调用方的集成验证。
实验读法与旧内容唯一归属
公开实验入口为 python3 labs/cloud-functions/run.py，本次2026-10-28已运行并生成 固定证据：10条测试通过，另1条契约违约负控按预期失败，总体exit=0。主工程覆盖 Function2、Stream4、Task2 条，独立Contract子工程覆盖生成verifier与StubRunner2条；9条观察行与完整Surefire报告分别保存。Kafka为真实单节点embedded KRaft、独有topic；Task为临时H2文件；provider为MockMvc、consumer stub为真实loopback HTTP。没有连接用户现有数据库或消息服务。
未运行清单（不得外推）：Rabbit binder、双binder同时运行、Kafka Streams/Pulsar binder及schema registry服务、Function web与云/gRPC适配、Task Batch事件桥和经消息启动、跨进程任务锁/TTL过期/崩溃恢复、事务性Kafka路径与外部数据库组合、Contract消息契约与REMOTE制品解析、AOT/Native。新增模块的配置与源码推演在本页单列；公开实验10条通过不能抵扣这些运行验证，也不能推出生产容量、多节点复制/ISR或跨故障域可靠性。
旧内容唯一归属，只回指不复制：
| 旧页（相对链接） | 保留什么 | 本页只补什么 |
|---|---|---|
| MQ生产存储消费可靠性 | 七个成功点、复制/ISR/HW、事务 EOS、幂等 producer | Stream 发送/事务配置如何映射到 produce 确认 |
| Kafka消费位点与重平衡 | 三种进度、提交源码、重平衡与优雅停机 | ackMode 到提交前沿的映射 |
| 重试、死信、延迟消息与消息积压 | 错误分类、三条重试路径、DLX/TTL、积压算例 | binder maxAttempts/enableDlq 的客户端机制与边界 |
| Config与Bus：配置读取、传播与版本回滚 | Bus 寻址、事件入口、故障链观察 | 权威转移：Bus 寻址… 一节（原 :826）的 startOffset/组语义与 部分实例失联与修复责任 一节（原 :950）的 max-attempts 语义归本页；U34 保留 Bus 叙事 |
| ThreadPoolExecutor：阻塞队列与拒绝策略 | 停止接纳→有界等待→处置未开始→协调在途；Future 不是任务记录 | Task 退出与 Stream 容器停止只回指 |
| FutureTask、CompletableFuture 与异步异常 | 结果/执行资源/业务状态三分；Future 不是事务 | 全页在 ack/退出码/stub 结果处引用此三分 |
| 调度执行：@Scheduled 语义、异常吞噬、关闭与观测 | @Scheduled 三组语义、单线程默认、异常吞噬 | 明确 @Scheduled ≠ Spring Cloud Task；Batch/Quartz 回指 U46 |
| Boot集成：自动配置条件与替换 | 消息四种接线与确认边界、容量与版本 | 区分 starter 自动配置与 Stream binder/binding 生命周期 |
| 远程客户端与容错组合：Feign 与 Resilience4j | Feign Contract/SpringMvcContract、首个请求 | 首段消歧：Feign Contract ≠ Spring Cloud Contract |
| Reactor核心：背压、调度与取消 | 订阅契约、背压、冷热流 | 只讲 Function 对 Flux/Mono 的适配边界 |
| 重复消费、幂等与顺序消息、Outbox、事务消息与最终一致性 | 重放身份、Inbox/Outbox、幂等键 | DLQ 回放与 Task/启动消息重复只回指，不重写幂等方案（U61） |
labs/config-bus 与 labs/cloud-functions 的 Kafka broker 配置、临时目录与 topic 都按“每次运行独有、定向清理”处理；Bus 实验里 spring.cloud.stream.bindings.springCloudBusInput.consumer.max-attempts=1 是真实绑定名，可作为本页“binding 属性键必须用推导名或 Bus 常量”的现成证据（Bus 常量 INPUT=springCloudBusInput，BusConstants.java:28，U34 审计报告 §1.4 的核实）。
实际观察分为四组：functionCatalog/functionReactive 验证组合、路由和流终止；streamCommit/streamDlq/streamNoDlq/streamBridge 验证处理阻塞前沿、重试耗尽后DLQ或恢复处置以及独立broker读回；taskRepository/taskSingleInstance 验证0/1/0的三次执行、持久查询与独立线程竞争；contractStubRunner 验证200/404。另有生成verifier的正控及BROKEN负控报告，不能只数观察行判断测试是否齐全。尤其 streamNoDlq 中失败业务数0而位点推进到下一条，正是“提交不等于业务成功”的反例。
K01–K09 的定位方式：参与者、问题与输入输出在各主节引入；完整 POM/应用/配置在“从依赖到 binding 的完整接线”“独立 catalog 的调用与可信路由”“Task 的完整启动与失败输入”“Contract 的生成与消费示例”，并分别列出默认、显式值、生效时点与单位。固定 source-excerpt 及就近解释支撑 K03/K09；binding 键写错、提前确认、不可重试载荷、路由越界及契约漂移等反例和诊断步骤支撑 K04–K06。闭卷题/答案与回查对应 K07；“容量、重试预算与恢复责任”比较两种重试位置、Task/Batch 恢复职责及 EOL 契约工具迁移，支撑 K08。新增教学代码与纸面算例不冒充上游源码或本次实测；运行证据仍以公开实验列出的实际断言为准。
用法来源可从同一固定版本的官方目录回查：Stream binding 属性、多 binder、Task 功能与配置、Contract 消息契约。25 个知识点 ID 与本页 H2 的对应（供登记使用）：
| 知识点 ID | H2 节 | 实跑/推演 |
|---|---|---|
| u36-stream-binder-model | Stream 应用模型与 binder SPI | 命令式Kafka真实binding；SPI其他实现推演 |
| u36-stream-functional-binding | functional model、functionDefinition 与 StreamBridge | 消费与StreamBridge实跑；输入绕过告警推演 |
| u36-stream-ack-commit | ackMode 与提交前沿 | 非事务RECORD阻塞/完成实跑；BATCH/MANUAL推演 |
| u36-stream-retry-dlq | 重试、DLQ 与客户端边界 | 3次耗尽、DLQ/无DLQ及后续处理实跑；事务路径未运行 |
| u36-stream-start-position | 起点、组与广播/竞争 | 显式新组earliest用于实验；匿名/已有组/reset推演 |
| u36-stream-multi-binder | 多 binder 与默认 binder | 仅一个Kafka binder实跑；双binder和Rabbit推演 |
| u36-stream-error-handling | 发送侧与消费侧错误出口 | 默认失败恢复/DLQ实跑；自定义error handler与发送故障推演 |
| u36-function-catalog | FunctionCatalog 与类型转换 | catalog查找、组合、路由实跑；其他转换条件推演 |
| u36-function-composition | 函数组合、路由与响应式函数 | 组合/路由/未知目标实跑；可信callback例子编译 |
| u36-function-reactive | 函数组合、路由与响应式函数 | catalog Flux直调实跑；Stream响应式装配与确认推演 |
| u36-function-web | Function web 适配 | 未运行 |
| u36-task-model | Task 生命周期与执行记录 | 成功/失败/新执行0/1/0持久查询实跑 |
| u36-task-repository | TaskRepository 与 schema、单实例锁 | H2文件库/两个线程独立Context竞争实跑；跨进程和TTL到期未运行 |
| u36-task-listener | Task 生命周期与执行记录 | AfterTask退出消息持久化实跑；ExitCodeEvent优先级推演 |
| u36-task-launch-boundary | Task 启动边界 | 推演 |
| u36-contract-dsl | Contract：DSL 与 verifier | Groovy生成verifier、READY正控与BROKEN负控实跑 |
| u36-contract-stubrunner | StubRunner 与消息契约边界 | 本机生成mapping与HTTP200/404实跑；REMOTE未运行 |
| u36-contract-messaging | StubRunner 与消息契约边界 | 未运行/推演 |
| u36-kafka-streams-binder | Kafka Streams binder：拓扑、状态与恢复身份 | 固定源码/完整配置与故障推演，未运行 |
| u36-pulsar-binder | Pulsar binder：订阅、schema 与确认差异 | 固定源码/完整配置与故障推演，未运行 |
| u36-schema-registry | Schema Registry：版本登记与兼容不是一件事 | 固定源码/完整配置与故障推演，未运行 |
| u36-function-adapters | Function adapters：同一函数的传输、装载与流程接点 | 固定源码/完整配置与故障推演，未运行 |
| u36-task-batch-bridge | Task与Batch/Stream桥接：一条运行记录不等于一个Job或一条消息 | 固定源码/完整配置与故障推演，未运行 |
| u36-framework-vs-broker | 框架词义 vs broker 实际保证 | 源码证据汇总 |
| u36-evidence | 实验读法与旧内容唯一归属 | 10条通过、1条契约预期失败、9条观察行与SHA证据汇总 |
闭卷验收
本页 37 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。