使用 Spring 集成组件在 2 个 JMS 队列之间关联消息

Question

我有 2 个 JMS 队列，我的应用程序使用Jms.messageDrivenChannelAdapter(...)组件订阅了这两个队列。

第一个队列接收类型Paid的消息。第二个队列接收类型Reversal的消息。

业务场景定义类型为Paid和类型Reversal的消息之间的关联。

Reversal应等待Paid才能进行处理。

如何通过 Spring 集成实现这种"等待"模式？

是否可以在 2 个 JMS 队列之间关联消息？

Answer 1

请参阅有关聚合器的文档。

聚合器使用某种关联策略关联消息，并根据某种发布策略发布组。

聚合器通过关联和存储一组相关消息来组合它们，直到该组被视为完整。此时，聚合器通过处理整个组来创建单个消息，并将聚合的消息作为输出发送。

默认情况下，输出有效负载是分组消息有效负载的列表，但您可以提供自定义输出处理器。

编辑

@SpringBootApplication
public class So55299268Application {
public static void main(String[] args) {
SpringApplication.run(So55299268Application.class, args);
}
@Bean
public IntegrationFlow in1(ConnectionFactory connectionFactory) {
return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(connectionFactory)
.destination("queue1"))
.channel("aggregator.input")
.get();
}
@Bean
public IntegrationFlow in2(ConnectionFactory connectionFactory) {
return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(connectionFactory)
.destination("queue2"))
.channel("aggregator.input")
.get();
}
@Bean
public IntegrationFlow aggregator() {
return f -> f
.aggregate(a -> a
.correlationExpression("headers.jms_correlationId")
.releaseExpression("size() == 2")
.expireGroupsUponCompletion(true)
.expireGroupsUponTimeout(true)
.groupTimeout(5_000L)
.discardChannel("discards.input"))
.handle(System.out::println);
}
@Bean
public IntegrationFlow discards() {
return f -> f.handle((p, h) -> {
System.out.println("Aggregation timed out for " + p);
return null;
});
}
@Bean
public ApplicationRunner runner(JmsTemplate template) {
return args -> {
send(template, "one", "two");
send(template, "three", null);
};
}
private void send(JmsTemplate template, String one, String two) {
template.convertAndSend("queue1", one, m -> {
m.setJMSCorrelationID(one);
return m;
});
if (two != null) {
template.convertAndSend("queue2", two, m -> {
m.setJMSCorrelationID(one);
return m;
});
}
}
}

和

GenericMessage [payload=[two， one]， headers={jms_redelivered=false， jms_destination=queue：//queue1， jms_correlationId=one， id=784535fe-8861-1b22-2cfa-cc2e67763674， priority=4， jms_timestamp=1553290921442， jms_messageId=ID：Gollum2.local-55540-1553290921241-4：1：3：1：1， timestamp=1553290921457}]

2019-03-22 17：42：06.460 信息 55396 --- [ask-scheduler-1] o.s.i.a.聚合消息处理程序：具有相关键的过期消息组[三]

聚合超时了三个