工程学习台

主题

主要章节

Spring Integration：通道语义、重试与错误通道的可恢复消息流 ​

Spring Integration 是独立项目，不是 Framework "Integration" 章节的别名：Framework 的集成能力（JMS 邮件/OXM/JMX/数据缓冲）归 U49，本页讲的是 message channel、端点与 advice 组成的消息流。本票固定组合 Spring Integration 7.0.7（Boot 4.0.9 BOM 钉版，spring-messaging/Framework 7.0.10 同链）、Liberica JDK 21.0.11，以进程内消息流实跑：无外部服务与持久化资源，输入在实验入口。与外部代理/重试的通用预算模型回查 HTTP 重试边界，与业务幂等回查重复消费与幂等。

通道语义：谁在哪个线程执行，决定错误去哪 ​

DirectChannel 在发送方线程内同步调用订阅者：处理器抛出的异常被包装成 MessageDeliveryException（携带 failedMessage 与 cause）抛给发送方，不进错误通道——实验观察到发送线程与处理线程为同一线程。这是文档明确的边界：消息式错误处理只适用于"在 TaskExecutor 内执行的 Spring Integration 任务"（QueueChannel 的轮询消费者、ExecutorChannel、poller 任务）。两条铁律：想要调用方感知失败，用 DirectChannel 并 try/catch；想要失败进错误通道，把消费放到异步侧（队列+轮询），并把 errorChannel 头指到自己的通道。

QueueChannel 是缓冲通道：send 入队、消费者线程 take/poll。固定源码里 send/receive 的超时语义一目了然：

原码java
	@Override
	protected boolean doSend(Message<?> message, long timeout) {
		Assert.notNull(message, "'message' must not be null");
		try {
			if (this.queue instanceof BlockingQueue) {
				BlockingQueue<Message<?>> blockingQueue = (BlockingQueue<Message<?>>) this.queue;
				if (timeout > 0) {
					return blockingQueue.offer(message, timeout, TimeUnit.MILLISECONDS);
				}
				if (timeout == 0) {
					return blockingQueue.offer(message);
				}
				blockingQueue.put(message);
				return true;
			}
			else {
				try {
					return this.queue.offer(message);
				}
				finally {
					this.queueSemaphore.release();
				}
			}
		}
		catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			return false;
		}
	}

	@Override
	@Nullable
	protected Message<?> doReceive(long timeout) {
		try {
			if (timeout > 0) {
				if (this.queue instanceof BlockingQueue) {
					return ((BlockingQueue<Message<?>>) this.queue).poll(timeout, TimeUnit.MILLISECONDS);
				}
				else {
					return pollNonBlockingQueue(timeout);
				}
			}
			if (timeout == 0) {
				return this.queue.poll();
			}

			if (this.queue instanceof BlockingQueue) {
				return ((BlockingQueue<Message<?>>) this.queue).take();
			}
			else {
				Message<?> message = this.queue.poll();
				while (message == null) {
					this.queueSemaphore.tryAcquire(50, TimeUnit.MILLISECONDS); // NOSONAR ok to ignore result
					message = this.queue.poll();

timeout=0 立即 offer/poll，timeout>0 限时阻塞，timeout<0 无限等待——容量满时 send 阻塞或失败，这正是背压的入口；但内存队列没有持久化：进程死了队列里的消息就没了，需要"可恢复"语义时要用持久化队列或外部 broker（回指对应课程），Spring Integration 的价值在流的结构，不在替你持久化。

变换与端点：@Transformer/@ServiceActivator 组装流 ​

@Transformer(inputChannel=..., outputChannel=...) 把方法注册为变换端点；@ServiceActivator 注册终(end)点，方法返回值可路由到 outputChannel。实验用 @Transformer 把 CSV 字符串变成列表、用 @ServiceActivator 完成重试与恢复链路——注解端点的线程模型仍由通道类型决定：csvChannel 是 DirectChannel，变换在发送方线程内完成。

重试：RequestHandlerRetryAdvice 与 Framework 7 的 RetryTemplate ​

7.0.7 的 RequestHandlerRetryAdvice 不再依赖 spring-retry，而是 Framework 7 的 org.springframework.core.retry.RetryTemplate。固定源码里能看到完整分支：

原码java
	protected @Nullable Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
		IntegrationRetryCallback retryCallback = new IntegrationRetryCallback(message, callback, target);
		AttributeAccessor retryContext = ErrorMessageUtils.getAttributeAccessor(message, message);
		try {
			if (this.stateKeyFunction == null) {
				return this.retryTemplate.<@Nullable Object>execute(retryCallback);
			}

			return statefulRetry(retryCallback, this.stateKeyFunction);
		}
		catch (RetryException ex) {
			if (this.recoveryCallback != null) {
				return this.recoveryCallback.recover(retryContext, ex);
			}

			Throwable cause = ex.getCause();

			if (cause instanceof MessagingException messagingException) {
				if (messagingException.getFailedMessage() == null) {
					throw new MessagingException(message, "Failed to invoke handler", ex);
				}
				throw messagingException;
			}
			else if (cause instanceof ThrowableHolderException throwableHolderException) {
				throw throwableHolderException;

无 stateKeyFunction 时走无状态 retryTemplate.execute；配置了 stateKey 时进入进程内有状态重试（缓存 RetryState 并按 backoff sleep——只对同进程有效，跨实例重试需要持久化状态）。RetryException 之后的分支决定失败去向：有 recoveryCallback 走恢复，否则把 cause 重新包装抛出、交给调用方或错误通道。注意语义细节：Framework 7 的 RetryPolicy.builder().maxRetries(2) 在本组合实测为首次执行 + 2 次重试共 3 次 attempts（实验计数 3），delay 是重试间隔——版本升级时这个计数语义要重新核对。

重试耗尽之后：ErrorMessageSendingRecoverer 与恢复责任 ​

耗尽后的恢复由 recoveryCallback 决定。ErrorMessageSendingRecoverer（官方 RecoveryCallback 实现）把失败包装成 ErrorMessage（payload 是异常、携带 failedMessage）发到指定通道——实验观察到 always-bad 的消息经恢复通道落地，主流程随后"修复"payload 重新投递并成功。这条链路的教学点不在 API，而在恢复责任分层：框架负责"把失败变成可消费的错误消息"；判断失败原因、修复并重新投递是应用层的事；重复投递的幂等由业务协议保证（回查重复消费与幂等），Integration 不提供去重。

错误通道：默认 errorChannel 与 advice 的失败路由 ​

默认 errorChannel 由框架创建，是带 LoggingHandler（ERROR 级）订阅者的 PublishSubscribeChannel；poller 任务的异常经 MessagePublishingErrorHandler 进入它。要自定义恢复，可以定义自己的 errorChannel（如 QueueChannel(500)）或按消息头的 errorChannel 路由——实验用消息头把 poison 消息的错误定向到自己的错误队列。advice 层还有更细的路由：ExpressionEvaluatingRequestHandlerAdvice 在处理器成功/失败时执行 SpEL：

原码java
	protected @Nullable Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
		try {
			Object result = callback.execute();
			if (this.onSuccessExpression != null) {
				evaluateSuccessExpression(message);
			}
			return result;
		}
		catch (RuntimeException e) {
			Exception actualException = unwrapExceptionIfNecessary(e);
			if (this.onFailureExpression != null) {
				Object evalResult = evaluateFailureExpression(message, actualException);
				if (this.returnFailureExpressionResult) {
					return evalResult;
				}
			}
			if (!this.trapException) {
				if (e instanceof ThrowableHolderException) { // NOSONAR
					throw e;
				}
				else {
					throw new ThrowableHolderException(actualException); // NOSONAR lost stack trace
				}
			}
			return null;

onFailureExpression 的求值结果不直接作为消息发送，而是装进 MessageHandlingExpressionEvaluatingAdviceException 再包成 ErrorMessage 发到 failureChannel——实验从异常包装里取回 getEvaluationResult() 得到 recorded:bad；trapException=true 时原始异常不再穿透到调用方。这套机制适合"失败留痕 + 继续流程"，不适合替代错误通道的恢复语义。

可恢复消息流：把各件拼起来 ​

一条可恢复链路的骨架：入站（队列+轮询，异步侧消费）→ 变换 → 带 retry advice 的服务 → 耗尽后经 recovery 进错误队列 → 人工/调度修复 → 重新投递。实验逐段验证了它，并留下三个必须显式回答的问题：消息在哪一层持久化（本实验：内存队列，进程死即丢）；重试是进程内有状态还是跨实例（本组合：进程内）；重新投递的幂等谁负责（应用协议）。 answered 之前不要把"有错误通道"当成"有恢复能力"。

观测、排障与恢复 ​

症状先查什么本票依据
DirectChannel 的异常"直接抛回调用方"这是同线程语义；错误通道只接异步侧失败directChannel 观察
错误消息没进 errorChannel消息头 errorChannel 优先于全局 bean；确认头是否存在queueChannel 观察
重试次数与配置不符Framework 7 maxRetries 语义（首次+重试），版本升级复核retry 观察
failureChannel 收到的是异常包装表达式结果在 getEvaluationResult()advice 观察
重启后队列消息丢失内存队列无持久化；持久化属外部组件通道语义

容量与取舍 ​

DirectChannel 同线程最简单也最快，但调用方承担全部失败处理；QueueChannel 引入线程切换与容量上限，换来生产/消费解耦和错误通道语义，代价是进程内缓冲的易失性与 maxCapacity 时的阻塞/丢弃决策。重试 advice 的 attempts 与 backoff 在消息流里是"调用侧预算"，与 HTTP 客户端重试、下游重试叠加时仍是乘法关系（回查 HTTP 重试边界），要在链路上取一个总预算 owner。进程内消息流的多实例扩展：队列不共享，每实例各消费各的；要共享就得换持久化存储或 broker，随之而来的是可见性与幂等协议。

条件变化与未运行边界 ​

未运行：JMS/HTTP/Mail 等真实入站出站适配器、跨进程/持久化通道（JDBC/Redis backing）、ExecutorChannel 与并发轮询、锁与共享队列的多实例语义、生产容量。重试次数语义（首次+maxRetries）为本组合 7.0.7 观察，升级需复核。stateKeyFunction 有状态重试只验证了实现存在，未运行其跨消息行为。教材"已讲清"不代表学习者掌握。

闭卷验收 ​

本页 12 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。

来源与版本 ​

来源版本用途
QueueChannel.javav7.0.7缓冲通道超时语义（固定摘录）
RequestHandlerRetryAdvice.javav7.0.7重试与恢复分支（固定摘录）
ExpressionEvaluatingRequestHandlerAdvice.javav7.0.7失败表达式路由（固定摘录）
错误处理参考（源 error-handling.adoc）7.0.7默认 errorChannel/MessagePublishingErrorHandler 语义
Framework 7 RetryPolicyv7.0.10重试策略 builder 语义

访问与运行日期：2026-10-31。源码片段为 v7.0.7 tag 的连续行段，整文件 SHA-256 随每段元数据登记；运行版本以证据 JSON 为准。

工程学习台　阅读讲解，学后自主自测。记录保存在本机浏览器。