java - 使用 Java Config 的 Spring 集成和 JMS

Question

我对 Spring Integration 还很陌生，想要实现一个非常简单的场景，该场景目前仅由 Spring Boot 应用程序和 @JmsListener 注释的 MDB 方法完成(在方法中硬编码)。

1. 从 JMS 队列(即 Websphere MQ)检索消息
2. 检查 header 字段并根据其值路由到特定服务方法
3. 将消息存储在数据库(即 MongoDB)中
4. 如果出现错误，请将错误消息存储在数据库的另一个集合中

有人可以为我提供此类场景配置的示例吗？我更喜欢 Java 配置，但 xml 也可以。不幸的是，我无法从 Spring 集成示例中获得完成此任务的知识。

提前致谢

Answer 1

我希望这是不言自明的并且足以让您开始......

@SpringBootApplication
public class So48223952Application {

    public static void main(String[] args) {
        SpringApplication.run(So48223952Application.class, args).close();
    }

    @Bean
    public ApplicationRunner runner(JmsTemplate template) {
        return args -> {
            template.convertAndSend("foo", "sendingFoo", m -> {
                m.setStringProperty("myHeader", "foo");
                return m;
            });
            template.convertAndSend("foo", "sendingBar", m -> {
                m.setStringProperty("myHeader", "bar");
                return m;
            });
            Thread.sleep(10_000);
        };
    }

    @Bean
    public IntegrationFlow flow(ConnectionFactory connectionFactory) {
        return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(connectionFactory)
                        .destination("foo"))
                .channel(MessageChannels.publishSubscribe("pubsub"))
                .route("headers['myHeader']",
                        m -> m.channelMapping("foo", "fooChannel")
                              .channelMapping("bar", "barChannel"))
                .get();
    }

    @Bean
    public IntegrationFlow toMongo() {
        return IntegrationFlows.from("pubsub")
                .<String, String>transform(p -> "Sending to db " + p)
                .handle(System.out::println) // store in DB here
                .get();
    }

    @Bean
    public IntegrationFlow foo() {
        return IntegrationFlows.from("fooChannel")
                .<String, String>transform(p -> "on fooChannel " + p)
                .handle(System.out::println)
                .get();
    }

    @Bean
    public IntegrationFlow bar() {
        return IntegrationFlows.from("barChannel")
                .<String, String>transform(p -> "on barChannel " + p)
                .handle(System.out::println)
                .get();
    }

}

结果:

GenericMessage [payload=on fooChannel sendingFoo, headers={jms_redelivered=false, myHeader=foo, jms_destination=queue://foo, id=ea65c71e-3702-88aa-fa07-fe0e53ec7539, priority=4, jms_timestamp=1515771377035, jms_messageId=ID:gollum.local-62392-1515771376836-4:2:1:1:1, timestamp=1515771377049}]
GenericMessage [payload=Sending to db sendingFoo, headers={jms_redelivered=false, myHeader=foo, jms_destination=queue://foo, id=26967f68-b2ad-a0f6-df62-5e8387f345f7, priority=4, jms_timestamp=1515771377035, jms_messageId=ID:gollum.local-62392-1515771376836-4:2:1:1:1, timestamp=1515771377049}]
GenericMessage [payload=on barChannel sendingBar, headers={jms_redelivered=false, myHeader=bar, jms_destination=queue://foo, id=6609a77c-55aa-9b84-49f6-da915b5d1734, priority=4, jms_timestamp=1515771377042, jms_messageId=ID:gollum.local-62392-1515771376836-4:3:1:1:1, timestamp=1515771377052}]
GenericMessage [payload=Sending to db sendingBar, headers={jms_redelivered=false, myHeader=bar, jms_destination=queue://foo, id=6b81ae82-4b2d-9d68-bbee-31a24e407565, priority=4, jms_timestamp=1515771377042, jms_messageId=ID:gollum.local-62392-1515771376836-4:3:1:1:1, timestamp=1515771377052}]

编辑

带有错误处理...

@SpringBootApplication
public class So48223952Application {

    public static void main(String[] args) {
        SpringApplication.run(So48223952Application.class, args).close();
    }

    @Bean
    public ApplicationRunner runner(JmsTemplate template) {
        return args -> {
            template.convertAndSend("foo", "sendingFoo", m -> {
                m.setStringProperty("myHeader", "foo");
                return m;
            });
            template.convertAndSend("foo", "sendingBar", m -> {
                m.setStringProperty("myHeader", "bar");
                return m;
            });
            Thread.sleep(10_000);
        };
    }

    @Bean
    public IntegrationFlow flow(ConnectionFactory connectionFactory) {
        return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(connectionFactory)
                        .destination("foo")
                        .errorChannel("errors"))
                .channel(MessageChannels.publishSubscribe("pubsub"))
                .route("headers['myHeader']",
                        m -> m.channelMapping("foo", "fooChannel")
                              .channelMapping("bar", "barChannel"))
                .get();
    }

    @Bean
    public IntegrationFlow toMongo() {
        return IntegrationFlows.from("pubsub")
                .<String, String>transform(p -> "Sending to db " + p)
                .handle(System.out::println) // store in DB here
                .get();
    }

    @Bean
    public IntegrationFlow foo() {
        return IntegrationFlows.from("fooChannel")
                .<String, String>transform(p -> "on fooChannel " + p)
                .handle(System.out::println)
                .get();
    }

    @Bean
    public IntegrationFlow bar() {
        return IntegrationFlows.from("barChannel")
                .<String, String>transform(p -> "on barChannel " + p)
                .handle(m -> {
                    throw new RuntimeException("error testing");
                })
                .get();
    }

    @Bean
    public IntegrationFlow errorFlow() {
        return IntegrationFlows.from("errors")
                .handle(m -> {
                    MessagingException me = (MessagingException) m.getPayload();
                    System.out.println("Message: " + me.getFailedMessage() + "\nFailed with "
                            + me.getCause().getMessage());
                })
                .get();
    }

}

和

GenericMessage [payload=on fooChannel sendingFoo, ...
GenericMessage [payload=Sending to db sendingFoo, ...
Message: GenericMessage [payload=on barChannel sendingBar, ...
Failed with error testing