我希望这是不言而喻的,足以让你开始。。。
@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)
.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:
GenericMessage [payload=Sending to db sendingFoo, headers={jms_redelivered=false, myHeader=foo, jms_destination=queue:
GenericMessage [payload=on barChannel sendingBar, headers={jms_redelivered=false, myHeader=bar, jms_destination=queue:
GenericMessage [payload=Sending to db sendingBar, headers={jms_redelivered=false, myHeader=bar, jms_destination=queue:
编辑
使用错误处理。。。
@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)
.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