我正在尝试了解聚合器的基础知识。以下是我尝试实现的用例：
1) 从队列中读取消息（订单详情）。
<?xml version="1.0" encoding="UTF-8"?>
<order xmlns="http://www.example.org/orders">
<orderItem>
<isbn>12333454443</isbn>
<quantity>4</quantity>
</orderItem>
<orderItem>
<isbn>545656777</isbn>
<quantity>50</quantity>
</orderItem>
..
..
</order>
一条订单消息将包含多个orderItem。我们可以期待队列中有数百条订单消息。
2) 最终结果 ::
a) 每个订单项都应该写入一个文件。
b) 4 个这样的文件应该被写入一个唯一的文件夹。
举个例子，假设我们收到了两条订单消息——每条都包含三个orderitem。
所以我们需要创建 2 个文件夹：
在“文件夹 1”中，应该有 4 个文件（每个文件中有 1 个订单项）
在“文件夹 2”中，应该有 2 个文件（每个文件中有 1 个订单项）。在这里为简单起见，我们假设没有更多的订单消息出现，我们可以在 5 分钟后写入。
执行：
- 我能够从队列（websphere MQ）中读取消息并成功解组消息。
- 使用拆分器根据订单项计数拆分消息。
- 使用 Aggregator 将消息分组为 4 大小。
根据我的理解，我无法让聚合器工作。
- 我在 4 个orderitem时推送一个订单，消息被正确聚合。
- 我推送一个带有 5 个orderitem的订单，前 4 个被聚合，但最后一个被发送到丢弃通道。这是预期的，因为 MessageGroup 被释放，所以最后一条消息被丢弃。
- 我推送两个订单，每个订单包含 2 个orderitem。最后 2 个订单项被发送到丢弃通道。
关联策略是硬编码的（OrderAggregator.java），但上述情况应该有效。
需要有关如何实现此用例的指示，我可以将它们分组为 4 并写入唯一的文件夹。请注意，orderitem都是独立的书单，它们之间没有任何关系。
下面是配置。
spring-bean.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans">
<int:channel id="mqInbound"/>
<int:channel id="item"/>
<int:channel id="itemList"/>
<int:channel id="aggregatorDiscardChannel"/>
<int-jms:message-driven-channel-adapter id="jmsIn"
channel="mqInbound"
destination="requestQueue"
message- converter="orderMessageConverter"/>
<int:splitter input-channel="mqInbound" output-channel="item" expression="payload.orderItem"/>
<int:chain id="aggregateList" input-channel="item" output-channel="itemList" >
<int:header-enricher>
<int:header name="sequenceSize" expression="4" overwrite="true"/>
</int:header-enricher>
<int:aggregator correlation-strategy="orderAggregator" correlation-strategy-method="groupOrders" discard-channel="aggregatorDiscardChannel" />
</int:chain>
<int:service-activator input-channel="itemList" ref="displayAggregatedList" method="display"/>
<int:service-activator input-channel="aggregatorDiscardChannel" ref="displayAggregatedList" method="displayDiscarded"/>
<bean id="orderAggregator" class="com.samples.Aggregator.OrderAggregator"/>
<bean id="displayAggregatedList" class="com.samples.Aggregator.DisplayAggregatedList"/>
...
....
</beans>
OrderAggregator.java
public class OrderAggregator {
@Aggregator
public List<OrderItemType> sendList(List<OrderItemType> orderItemTypeList) {
return orderItemTypeList;
}
@CorrelationStrategy
public String groupOrders( OrderItemType orderItemType) {
return "items";
}
}
DisplayAggregatedList.java
public class DisplayAggregatedList {
public void display(List <OrderItemType> orderItemTypeList) {
System.out.println("######## Display Aggregated ##############");
for(OrderItemType oit : orderItemTypeList) {
System.out.println("### Isbn :" + oit.getIsbn() + ":: Quantity :" + oit.getQuantity());
}
}
public void displayDiscarded(Message<?> message) {
System.out.println("######## Display Discarded ##############" + message);
}
}