服务端开发

Appearance

🚀 消息驱动批处理：Spring Batch与Spring Integration集成指南 ​

批处理新视角：想象一下，当文件像快递包裹一样到达指定位置时，系统自动拆包处理并反馈结果——这就是消息驱动批处理的魅力！

一、传统批处理启动方式的局限 ​

在Spring Batch中启动任务通常有两种方式：

kotlin
// 1️⃣ 命令行启动（适合脚本调用）
// 语法：java -jar CommandLineJobRunner jobName param1=value1

// 2️⃣ 编程式启动（适合Web应用）
@Autowired
private lateinit var jobLauncher: JobLauncher

fun launchJob() {
    val jobParameters = JobParametersBuilder()
        .addString("input", "data.csv")
        .toJobParameters()

    jobLauncher.run(job, jobParameters)
}

传统方式的痛点

  • 硬编码依赖：文件路径/参数需要预先确定
  • 缺乏事件响应：无法实时响应文件到达等事件
  • 多源支持困难：同时处理FTP/SFTP/邮件附件等来源复杂

二、消息驱动批处理架构 ​

🌐 集成架构时序图 ​

🔑 核心组件解析 ​

kotlin
// [!code highlight:1-3,7-9]
// 消息转换器：文件→Job请求
class FileToJobRequestTransformer(
    private val job: Job,
    private val fileParamName: String = "input.file.name"
) {
    @Transformer
    fun transform(fileMessage: Message<File>): JobLaunchRequest {
        val params = JobParametersBuilder()
            .addString(fileParamName, fileMessage.payload.absolutePath)
            .toJobParameters()

        return JobLaunchRequest(job, params)
    }
}

// 作业启动网关
@Bean
fun jobLaunchingGateway(jobRepository: JobRepository) = JobLaunchingGateway(
    TaskExecutorJobLauncher().apply {
        setJobRepository(jobRepository)
        setTaskExecutor(SyncTaskExecutor())  
    }
)

TIP

SyncTaskExecutor警告： 使用同步执行器会阻塞消息线程直到作业完成，在高并发场景请考虑ThreadPoolTaskExecutor

三、全流程配置实战（Kotlin DSL） ​

🔧 集成配置示例 ​

kotlin
@Configuration
@EnableIntegration
class BatchIntegrationConfig {

    // 1. 文件监听器
    @Bean
    fun fileInputFlow() = IntegrationFlow.from(
        Files.inboundAdapter(File("/data/inbound"))
            .filter(SimplePatternFileListFilter("*.csv")),
        { it.poller(Pollers.fixedDelay(1000).maxMessagesPerPoll(1) }
    )
    .transform(fileToJobRequestTransformer()) // 2. 文件转Job请求
    .handle(jobLaunchingGateway())            // 3. 启动批处理作业
    .log(LoggingHandler.Level.INFO, "'Job launched: ' + payload")
    .get()

    // 转换器Bean
    @Bean
    fun fileToJobRequestTransformer() = FileToJobRequestTransformer(dataJob())

    // 批处理作业
    @Bean
    fun dataJob() = JobBuilder("csvProcessingJob")
        .start(stepBuilderFactory.get("processStep")
            .chunk<Person, Person>(100)
            .reader(personReader(null))
            .processor(personProcessor())
            .writer(jdbcBatchItemWriter())
            .build())
        .build()

    // 4. 动态文件读取器
    @Bean
    @StepScope
    fun personReader(
        @Value("#{jobParameters['input.file.name']}") filePath: String?
    ) = FlatFileItemReader<Person>().apply {
        setResource(FileSystemResource(filePath))
        setLineMapper(defaultLineMapper())
    }
}

IMPORTANT

关键配置说明：

  1. @StepScope确保每次作业执行创建新的Reader实例
  2. #{jobParameters[...]}实现运行时参数注入
  3. 文件路径参数名需与转换器中fileParamName一致

四、执行响应处理策略 ​

🔄 同步vs异步响应对比 ​

特性同步执行(SyncTaskExecutor)异步执行(ThreadPoolTaskExecutor)
响应时机作业完成后返回立即返回JobExecution
资源占用阻塞调用线程非阻塞，线程池处理
结果获取直接获得最终状态需通过JobExplorer查询状态
适用场景轻量任务/测试生产环境高并发

🔍 异步作业状态查询 ​

kotlin
@Service
class JobStatusService(
    private val jobExplorer: JobExplorer
) {
    fun getJobStatus(jobExecutionId: Long): String {
        val execution = jobExplorer.getJobExecution(jobExecutionId)
        return execution?.status?.name ?: "UNKNOWN"
    }
}

五、生产环境最佳实践 ​

⚠️ 安全增强配置 ​

kotlin

@Bean
fun securedJobLauncher(): JobLauncher {
    return object : SimpleJobLauncher() {
        override fun run(job: Job, jobParameters: JobParameters): JobExecution {
            validateParameters(jobParameters) // 自定义参数验证
            return super.run(job, jobParameters)
        }
    }
}

private fun validateParameters(params: JobParameters) {
    if (!params.containsKey("security.token"))
        throw JobSecurityException("缺少安全令牌")
}

📁 文件处理增强 ​

kotlin
// 文件处理拦截器
@Bean
fun fileProcessingInterceptor() = object : FileTransferringMessageHandler {
    override fun handleMessage(message: Message<*>) {
        val file = (message.payload as File)
        // 1. 病毒扫描
        if (virusScanner.detectThreats(file)) {
            quarantineFile(file)  // 隔离危险文件
        }
        // 2. 备份原始文件
        archiveService.backup(file)

        super.handleMessage(message)
    }
}

绝对路径安全警告

避免在参数中使用绝对路径！应通过配置中心动态获取路径：

kotlin
@Value("\${batch.input.dir}")
lateinit var inputDir: String

六、多源扩展方案 ​

🌈 支持多种输入源 ​

kotlin
@Bean
fun multiSourceIntegration() = IntegrationFlows.from(
    // FTP源
    Ftp.inboundAdapter(ftpSessionFactory())
        .remoteDirectory("/inbound")
        .regexFilter(".*\\.csv$"),
    { it.poller(Pollers.fixedRate(5000)) }
).and(
    // HTTP源
    Http.inboundChannelAdapter("/batch-input")
        .requestPayloadType(File::class.java)
).transform(fileToJobRequestTransformer())
.handle(jobLaunchingGateway())

七、调试与监控 ​

🔭 监控批处理事件 ​

kotlin
@EventListener
fun handleJobEvent(event: JobExecutionEvent) {
    when (event) {
        is JobExecutionStartingEvent ->
            logger.info("作业启动: ${event.jobExecution.jobId}")
        is JobExecutionFailedEvent ->
            logger.error("作业失败", event.cause)  
        is JobExecutionCompletedEvent ->
            metrics.trackDuration(event.duration)
    }
}

CAUTION

生产环境注意事项：

  1. 实现死信队列处理失败消息
  2. 设置文件处理超时机制
  3. 添加作业并发控制

结语：消息驱动优势总结 ​

✅ 事件驱动：实时响应文件到达等事件
✅ 解耦架构：批处理核心与触发机制分离
✅ 扩展灵活：轻松集成FTP/S3/邮件等输入源
✅ 资源优化：异步处理避免线程阻塞

kotlin
// 最终集成效果：一行命令启动完整处理流程
curl -X POST -F "[email protected]" http://yourapp/batch-input

通过本教程，您已掌握使用Spring Integration消息驱动Spring Batch作业的核心技能。现在就去创建您的第一个响应式批处理系统吧！