Spring Integration

Appearance

Spring Integration 错误处理指南 ​

引言：消息系统中的错误挑战 ​

NOTE

在松散耦合的消息系统中，错误处理面临独特挑战：生产者与消费者分离，异常无法通过传统调用栈传播。

Spring Integration 通过异步错误通道机制解决此问题：

一、同步 vs 异步错误处理 ​

1.1 同步处理（DirectChannel） ​

kotlin
@Bean
fun directChannel() = DirectChannel().apply {
    subscribe { message ->
        // 同步处理中异常会直接抛出
        throw RuntimeException("同步处理失败")
    }
}

// 调用示例
fun main() {
    try {
        directChannel().send(MessageBuilder.withPayload("test").build())
    } catch (ex: MessagingException) {
        // 可在此捕获异常
        println("捕获异常: ${ex.message}")
    }
}

1.2 异步处理（QueueChannel/ExecutorChannel） ​

kotlin
@Bean
fun queueChannel(): PollableChannel {
    return QueueChannel(10).apply {
        subscribe { message ->
            // 异常无法传播回发送者
            throw RuntimeException("异步处理失败")
        }
    }
}

CAUTION

异步环境中异常不能直接返回给发送方，必须通过错误通道机制处理

二、错误通道机制解析 ​

2.1 错误传播流程 ​

2.2 核心组件 ​

组件作用默认值
errorChannel全局错误通道PublishSubscribeChannel
LoggingHandler默认日志处理器ERROR级别
MessageHeaders.ERROR_CHANNEL消息头键"errorChannel"

三、配置错误处理通道 ​

3.1 自定义错误通道配置 ​

kotlin
@Configuration
class ErrorConfig {

    // 创建容量500的队列通道
    @Bean
    fun errorChannel(): PollableChannel {
        return QueueChannel(500)
    }

    // 注册全局错误处理器
    @Bean
    fun errorHandler(): MessageHandler {
        return object : AbstractReplyProducingMessageHandler() {
            override fun handleMessageInternal(message: Message<*>) {
                val exception = (message.payload as MessagingException).cause
                println("全局捕获异常: ${exception?.message}")
            }
        }
    }
}

TIP

使用@ServiceActivator注解可将方法绑定到错误通道：

kotlin
@ServiceActivator(inputChannel = "errorChannel")
fun handleError(error: ErrorMessage) {
    val exception = error.payload as MessagingException
    println("处理异常: ${exception.failedMessage}")
}

四、高级错误处理策略 ​

4.1 异常类型路由 ​

kotlin
@Bean
fun exceptionRouter(): MessageHandler {
    return ErrorMessageExceptionTypeRouter().apply {
        setChannelMapping(IllegalArgumentException::class.java.name, "illegalArgumentChannel")
        setChannelMapping(NullPointerException::class.java.name, "npeChannel")
    }
}

@ServiceActivator(inputChannel = "illegalArgumentChannel")
fun handleIllegalArgument(ex: MessagingException) {
    println("参数异常处理: ${ex.message}")
}

4.2 重试机制整合 ​

kotlin
@Bean
fun retryAdvice(): RequestHandlerRetryAdvice {
    return RequestHandlerRetryAdvice().apply {
        setRecoveryCallback(ErrorMessageSendingRecoverer())
        retryTemplate = RetryTemplate().apply {
            setRetryPolicy(SimpleRetryPolicy(3))
        }
    }
}

五、关键注意事项 ​

5.1 错误处理范围限制 ​

WARNING

错误通道仅处理在TaskExecutor中执行的异常，同步处理中的异常会直接抛出！

5.2 默认通道行为变化 ​

IMPORTANT

Spring Integration 5.4.3+ 中errorChannel默认设置requireSubscribers=true：

  • ✅ 防止无订阅者时静默忽略错误
  • ❌ 无订阅者时会抛出MessageDispatchingException

恢复旧行为需配置：

properties
spring.integration.channels.error.requireSubscribers=false

5.3 调度任务错误处理 ​

kotlin
@Scheduled(fixedRate = 5000)
fun scheduledTask() {
    // 异常会被包装为ErrorMessage
    throw RuntimeException("定时任务异常")
}

@Bean
fun taskScheduler(): TaskScheduler {
    return ThreadPoolTaskScheduler().apply {
        errorHandler = MessagePublishingErrorHandler()
    }
}

六、最佳实践总结 ​

  1. 通道选择策略：

    • 同步操作：使用DirectChannel+传统异常处理
    • 异步操作：必须配置错误通道

  2. 错误处理金字塔：

  3. 诊断增强：

    kotlin
    try {
    // 业务代码
} catch (ex: MessageHandlingException) {
    // 5.2+版本包含配置源信息
    println("配置资源: ${ex.source}")
    println("Bean定义: ${ex.beanDefinition}")
}

实际应用场景

电商订单处理系统：

七、常见问题解决 ​

Q1: 为什么异步处理中无法捕获异常？
A: 异步操作发生在不同线程，必须通过错误通道机制处理

Q2: 如何区分业务异常和系统异常？
A: 使用ErrorMessageExceptionTypeRouter按异常类型路由

Q3: 错误消息中包含哪些关键信息？

kotlin
@ServiceActivator(inputChannel = "errorChannel")
fun inspectError(error: ErrorMessage) {
    val original = error.originalMessage      // 原始消息
    val exception = error.payload            // 异常对象
    val timestamp = error.headers.timestamp  // 发生时间
}

通过本指南，您已掌握Spring Integration的错误处理机制。建议参考官方Error Handling Sample获取完整示例。