Spring Integration DSL Flow配置：实现单消息串行处理

要实现发布-订阅通道的串行消费（一次仅处理一条消息，完成后再接收下一条），核心是确保当前消费流的处理逻辑在单线程环境中执行，且每次仅获取一条消息进行处理。以下是具体实现方案：

核心思路 #

发布-订阅通道默认是广播式消息分发，消费线程由发送端或通道配置决定，可能存在并行处理的情况。要实现串行，需通过轮询器+单线程执行器或队列通道缓冲的方式，强制消费流按顺序处理消息。

方案一：直接配置轮询器与单线程执行器 #

修改现有buildFlow方法，在from方法中配置轮询器，指定每次轮询仅获取1条消息，并使用单线程执行器处理所有步骤：

// 先定义单线程执行器Bean（推荐）
@Bean
public Executor importTaskSingleThreadExecutor() {
    return Executors.newSingleThreadExecutor(r -> {
        Thread thread = new Thread(r);
        thread.setName("import-task-processing-thread");
        thread.setDaemon(true);
        return thread;
    });
}

protected IntegrationFlowDefinition<?> buildFlow() {
    return from(TaskService.TASKS_CHANNEL_NAME, 
            c -> c.poller(Pollers.fixedDelay(100)
                    .maxMessagesPerPoll(1) // 每次轮询仅取1条消息
                    .taskExecutor(importTaskSingleThreadExecutor()))) // 单线程执行处理逻辑
            .filter(Task.class, task -> ImportTaskData.TASK_TYPE.equals(task.getTaskType()))
            .enrichHeaders(headerEnricher -> headerEnricher.headerExpression(TASK_UUID_HEADER_KEY, "payload.uuid"))
            .transform(Task.class, task -> task.getTaskDataAs(ImportTaskData.class))
            .enrichHeaders(headerEnricher -> headerEnricher.headerExpression(TASK_RESULTS_DATA_ID_HEADER_KEY, "payload.resultsDataId"))
            .transform(Message.class, message -> loadAndTransformImportFile(message))
            .transform(Message.class, message -> parseDataRows(message))
            .<Pair<Workbook,List<ImportRowData>>>handle((payload, headers) -> performImport(headers, payload))
            .<Pair<Workbook,ImportTaskResults>>handle((payload, headers) -> processResults(headers, payload));
}

关键配置说明 #

• maxMessagesPerPoll(1)：限制每次轮询仅从发布-订阅通道获取1条消息，避免一次性拉取多条并行处理。
• 单线程执行器：确保所有消息处理步骤（过滤、转换、业务逻辑）都在同一个线程中串行执行，处理完当前消息后，轮询器才会获取下一条消息。

方案二：通过队列通道缓冲实现串行 #

先将发布-订阅通道的消息转发到队列通道（点对点通道，默认串行消费），再从队列通道轮询消费：

// 定义队列通道Bean
@Bean
public QueueChannel importTaskQueue() {
    return new QueueChannel();
}

protected IntegrationFlowDefinition<?> buildFlow() {
    // 第一步：从发布-订阅通道接收消息，过滤后转发到队列通道
    from(TaskService.TASKS_CHANNEL_NAME)
            .filter(Task.class, task -> ImportTaskData.TASK_TYPE.equals(task.getTaskType()))
            .channel("importTaskQueue");
    
    // 第二步：从队列通道轮询消费，默认串行处理
    return from("importTaskQueue", c -> c.poller(Pollers.fixedDelay(100)))
            .enrichHeaders(headerEnricher -> headerEnricher.headerExpression(TASK_UUID_HEADER_KEY, "payload.uuid"))
            .transform(Task.class, task -> task.getTaskDataAs(ImportTaskData.class))
            .enrichHeaders(headerEnricher -> headerEnricher.headerExpression(TASK_RESULTS_DATA_ID_HEADER_KEY, "payload.resultsDataId"))
            .transform(Message.class, message -> loadAndTransformImportFile(message))
            .transform(Message.class, message -> parseDataRows(message))
            .<Pair<Workbook,List<ImportRowData>>>handle((payload, headers) -> performImport(headers, payload))
            .<Pair<Workbook,ImportTaskResults>>handle((payload, headers) -> processResults(headers, payload));
}

关键配置说明 #

• 队列通道（QueueChannel）是点对点模式，同一时间只有一个消费者能获取消息，天然支持串行处理。
• 轮询器的fixedDelay(100)表示处理完当前消息后，间隔100ms再轮询下一条消息，可根据业务调整间隔时间。

注意事项 #

• 单线程执行器建议定义为Bean，避免每次创建新线程池，便于统一管理线程生命周期。
• 如果业务处理耗时较长，轮询间隔可适当调大，避免空轮询浪费资源。

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