使用Spring Integration是否必须部署Web服务？如何让批处理监听消息启动任务

核心问题解答 #

• 不需要将批处理应用改造为Web应用，Spring Integration完全支持在非Web的Spring上下文环境中运行，仅依靠内置的消息监听容器线程即可保持运行状态，不需要依赖Servlet容器。
• 你可以通过消息中间件解耦Web触发端和批处理执行端，保持原有各批处理应用上下文完全隔离的特性，和你现有架构的隔离性一致。

具体实现步骤（以RabbitMQ作为消息中间件为例，可替换为Kafka/RocketMQ等任意Spring Integration支持的消息组件） #

1. 批处理执行端改造（无需改Web类型） #

首先引入依赖（以Maven为例）：

<!-- Spring Integration RabbitMQ 支持 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-amqp</artifactId>
</dependency>
<!-- Spring Batch Integration 适配 -->
<dependency>
    <groupId>org.springframework.batch</groupId>
    <artifactId>spring-batch-integration</artifactId>
</dependency>

然后添加配置类，实现消息监听+自动触发作业：

@Configuration
@EnableIntegration
public class BatchJobListenerConfig {

    // 配置当前批处理应用监听的专属队列
    @Bean
    public Queue batchJobQueue() {
        return new Queue("batch1-job-queue", true);
    }

    // 配置AMQP入站通道适配器，监听队列消息
    @Bean
    public AmqpInboundChannelAdapter inboundChannelAdapter(ConnectionFactory connectionFactory, Queue batchJobQueue) {
        AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(
                SimpleMessageListenerContainer(connectionFactory, batchJobQueue)
        );
        adapter.setOutputChannel(jobLaunchingChannel());
        // 配置消息转换器，将队列中的JSON消息转为JobParameters对象
        adapter.setMessageConverter(new Jackson2JsonMessageConverter());
        return adapter;
    }

    @Bean
    public MessageChannel jobLaunchingChannel() {
        return new DirectChannel();
    }

    // 配置Spring Batch Integration提供的作业启动网关，收到消息自动触发作业
    @Bean
    @ServiceActivator(inputChannel = "jobLaunchingChannel")
    public JobLaunchingGateway jobLaunchingGateway(JobLauncher jobLauncher) {
        JobLaunchingGateway gateway = new JobLaunchingGateway(jobLauncher);
        // 可配置作业执行后返回状态到回调队列，按需开启
        // gateway.setOutputChannel(jobStatusReplyChannel());
        return gateway;
    }
}

原有启动类无需修改，还是保持web(WebApplicationType.NONE)配置即可，启动后消息监听容器会作为非守护线程运行，应用会一直保持监听状态不会自动退出。

2. Web触发端改造 #

引入和执行端相同的消息中间件Spring Integration依赖，添加消息发送配置：

@Configuration
@EnableIntegration
public class BatchJobTriggerConfig {

    @Bean
    public MessageChannel batch1JobTriggerChannel() {
        return new DirectChannel();
    }

    // 配置AMQP出站适配器，将消息发送到对应批处理的队列
    @Bean
    @ServiceActivator(inputChannel = "batch1JobTriggerChannel")
    public AmqpOutboundEndpoint outboundEndpoint(ConnectionFactory connectionFactory) {
        AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(rabbitTemplate(connectionFactory));
        endpoint.setRoutingKey("batch1-job-queue");
        return endpoint;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }
}

需要触发作业时，直接向通道发送携带JobParameters的消息即可：

@Autowired
private MessageChannel batch1JobTriggerChannel;

public void triggerBatch1Job(Map<String, Object> jobParams) {
    JobParameters parameters = new JobParametersBuilder()
            .addDate("triggerTime", new Date())
            // 添加自定义参数
            .toJobParameters();
    batch1JobTriggerChannel.send(MessageBuilder.withPayload(parameters).build());
}

方案优势 #

• 保持原有各批处理应用上下文完全隔离的特性，互不干扰
• 比原有脚本启动JVM的方式延迟更低，避免重复加载Spring上下文的开销
• 可依托消息中间件的持久化、重试、死信队列能力，提升触发可靠性
• 可方便扩展作业执行状态回传、执行日志上报等能力，无需修改核心架构

内容的提问来源于stack exchange，提问作者flywell