Java企业集成模式：管道和过滤器模式

1. 管道和过滤器模式

• 管道（Pipe）：消息传输的通道
• 过滤器（Filter）：消息处理组件（转换、验证、增强）
• 模式特点：
  • 每个过滤器独立、高内聚
  • 可灵活组合和编排
  • 支持并行处理（过滤器间无状态依赖时）

2. Spring Integration实现

// 定义管道和过滤器
@Bean
public IntegrationFlow myFlow() {
    return IntegrationFlow.from("inputChannel")
        .filter("true", f -> f.throwExceptionOnRejection(true))  // 过滤
        .transform(Transformers.objectToString())                 // 转换
        .enrichHeaders(h -> h.header("X-Timestamp", 
            new Date()::toString))                                // 增强
        .<String, String>transform(String::toUpperCase)           // 自定义转换
        .handle(System.out::println)                              // 处理
        .get();
}

// 消息驱动
@Bean
public IntegrationFlow fileFlow() {
    return IntegrationFlow
        .from(Files.inboundAdapter(new File("/input")), e -> e.poller(Pollers.fixedDelay(1000)))
        .transform(Transformers.fileToString())
        .split(s -> s.delimiters("\n"))
        .handle("myService", "processLine")
        .get();
}

3. Apache Camel实现

from("file:input?noop=true")           // 管道入口
    .unmarshal().csv()                   // 解组（过滤器）
    .split(body())                       // 分割（过滤器）
    .process(exchange -> {               // 处理（过滤器）
        String line = exchange.getIn().getBody(String.class);
        exchange.getIn().setBody(line.toUpperCase());
    })
    .to("jms:queue:output");            // 管道出口

4. 优势

• 松耦合：过滤器独立开发、独立部署
• 可复用：过滤器可跨不同管道使用
• 可测试：每个过滤器可单独测试
• 可扩展：插入新过滤器不影响现有流程

5. 注意事项

• 过多过滤器增加延迟
• 共享状态需线程安全
• 错误处理应在过滤器链中添加错误通道

return IntegrationFlow.from("input")
    .handle("service", "process", e -> e.advice(retryAdvice()))
    .channel(MessageChannels.publishSubscribe())
    .get();