使用协议适配器

DeepSeek V3 中英对照 Using Protocol Adapters

到目前为止展示的所有示例都说明了DSL如何通过使用Spring Integration编程模型来支持消息传递架构。然而，我们尚未进行任何实际的集成操作。实现集成需要访问通过HTTP、JMS、AMQP、TCP、JDBC、FTP、SMTP等协议访问远程资源，或访问本地文件系统。Spring Integration支持所有这些协议及更多。理想情况下，DSL应该为所有这些提供一流的支持，但实现所有这些并在Spring Integration添加新适配器时保持同步是一项艰巨的任务。因此，期望DSL能够持续跟进Spring Integration的发展。

因此，我们提供了高级API来无缝定义特定协议的消息传递。我们通过工厂模式、构建器模式以及lambda表达式来实现这一点。您可以将工厂类视为“命名空间工厂”，因为它们与具体协议特定的Spring Integration模块中的XML命名空间扮演着相同的角色。目前，Spring Integration Java DSL支持 Amqp、Feed、Jms、Files、(S)Ftp、Http、JPA、MongoDb、TCP/UDP、Mail、WebFlux 和 Scripts 命名空间工厂。以下示例展示了如何使用其中的三个（Amqp、Jms 和 Mail）：

@Bean
public IntegrationFlow amqpFlow() {
    return IntegrationFlow.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue()))
            .transform("hello "::concat)
            .transform(String.class, String::toUpperCase)
            .get();
}

@Bean
public IntegrationFlow jmsOutboundGatewayFlow() {
    return IntegrationFlow.from("jmsOutboundGatewayChannel")
            .handle(Jms.outboundGateway(this.jmsConnectionFactory)
                        .replyContainer(c ->
                                    c.concurrentConsumers(3)
                                            .sessionTransacted(true))
                        .requestDestination("jmsPipelineTest"))
            .get();
}

@Bean
public IntegrationFlow sendMailFlow() {
    return IntegrationFlow.from("sendMailChannel")
            .handle(Mail.outboundAdapter("localhost")
                            .port(smtpPort)
                            .credentials("user", "pw")
                            .protocol("smtp")
                            .javaMailProperties(p -> p.put("mail.debug", "true")),
                    e -> e.id("sendMailEndpoint"))
            .get();
}

前面的示例展示了如何使用“命名空间工厂”作为内联适配器声明。然而，你也可以在 @Bean 定义中使用它们，以使 IntegrationFlow 方法链更具可读性。

备注

在投入精力开发其他命名空间工厂之前，我们正在就这些命名空间工厂征求社区反馈。同时，我们也欢迎对我们接下来应优先支持哪些适配器和网关提出建议。

您可以在本参考手册的各个协议特定章节中找到更多 Java DSL 示例。

所有其他协议通道适配器都可以配置为通用 Bean，并连接到 IntegrationFlow，如下例所示：

@Bean
public QueueChannelSpec wrongMessagesChannel() {
    return MessageChannels
            .queue()
            .wireTap("wrongMessagesWireTapChannel");
}

@Bean
public IntegrationFlow xpathFlow(MessageChannel wrongMessagesChannel) {
    return IntegrationFlow.from("inputChannel")
            .filter(new StringValueTestXPathMessageSelector("namespace-uri(/*)", "my:namespace"),
                    e -> e.discardChannel(wrongMessagesChannel))
            .log(LoggingHandler.Level.ERROR, "test.category", m -> m.getHeaders().getId())
            .route(xpathRouter(wrongMessagesChannel))
            .get();
}

@Bean
public AbstractMappingMessageRouter xpathRouter(MessageChannel wrongMessagesChannel) {
    XPathRouter router = new XPathRouter("local-name(/*)");
    router.setEvaluateAsString(true);
    router.setResolutionRequired(false);
    router.setDefaultOutputChannel(wrongMessagesChannel);
    router.setChannelMapping("Tags", "splittingChannel");
    router.setChannelMapping("Tag", "receivedChannel");
    return router;
}