Spring集成-优先级聚合器

Question

我有以下应用程序要求：

• 从RabbitMq接收消息，然后根据一些更复杂的规则进行聚合，例如基于类型属性（具有预先给定的类型时间映射）和基于消息在队列中等待的现有时间（旧属性）

正如您在图中看到的一个用例：三条消息已经聚合并等待下一秒发布（因为当前速率为1msg/sec），但就在那时，MSG以id:10到达，并更新了aggregated 2，使其成为优先级最高的第一条消息。因此，在下一个选项中，我们不再发布聚合3，而是发布聚合2，因为它现在具有更高的优先级。

现在，问题是-我可以为此使用Spring集成聚合器吗，因为我不知道它是否支持聚合期间消息的优先级？我知道group pTimeout，但那只是调整单个消息组-不改变其他组的优先级。是否有可能使用MessageGroupStorepper在新MSG到达时按优先级调整所有其他聚合消息？

使现代化

我做了一些这样的实现——目前看来还可以——它在消息到达时聚合消息，而comparator则根据我的自定义逻辑对消息进行排序。

2021-01-18 13:52:05.277  INFO 16080 --- [   scheduling-1] ggregatorConfig$PriorityAggregatingQueue : POLL
2021-01-18 13:52:05.277  INFO 16080 --- [   scheduling-1] ggregatorConfig$PriorityAggregatingQueue : POLL
2021-01-18 13:52:05.277  INFO 16080 --- [   scheduling-1] ggregatorConfig$PriorityAggregatingQueue : POLL
2021-01-18 13:52:05.277  INFO 16080 --- [   scheduling-1] ggregatorConfig$PriorityAggregatingQueue : POLL

另外，这个有注释的doit方法是增加运行时轮询消息最大数量的正确方法吗？

@Bean
    public MessageChannel aggregatingChannel(){
        return new QueueChannel(new PriorityAggregatingQueue<>((m1, m2) -> {//aggr here},
                Comparator.comparingInt(x -> x),
                (m) -> {
                    ExampleDTO d = (ExampleDTO) m.getPayload();
                    return d.getId();
                }
        ));
    }

    class PriorityAggregatingQueue<K> extends AbstractQueue<Message<?>> {
        private final Log logger = LogFactory.getLog(getClass());
        private final BiFunction<Message<?>, Message<?>, Message<?>> accumulator;
        private final Function<Message<?>, K> keyExtractor;
        private final NavigableMap<K, Message<?>> keyToAggregatedMessage;

        public PriorityAggregatingQueue(BiFunction<Message<?>, Message<?>, Message<?>> accumulator,
                                        Comparator<? super K> comparator,
                                        Function<Message<?>, K> keyExtractor) {
            this.accumulator = accumulator;
            this.keyExtractor = keyExtractor;
            keyToAggregatedMessage = new ConcurrentSkipListMap<>(comparator);
        }

        @Override
        public Iterator<Message<?>> iterator() {
            return keyToAggregatedMessage.values().iterator();
        }

        @Override
        public int size() {
            return keyToAggregatedMessage.size();
        }

        @Override
        public boolean offer(Message<?> m) {
            logger.info("OFFER");
            return keyToAggregatedMessage.compute(keyExtractor.apply(m), (k,old) -> accumulator.apply(old, m)) != null;
        }

        @Override
        public Message<?> poll() {
            logger.info("POLL");
            Map.Entry<K, Message<?>> m = keyToAggregatedMessage.pollLastEntry();
            return m != null ? m.getValue() : null;
        }

        @Override
        public Message<?> peek() {
            Map.Entry<K, Message<?>> m = keyToAggregatedMessage.lastEntry();
            return m!= null ? m.getValue() : null;
        }
    }

//    @Scheduled(fixedDelay = 10*1000)
//    public void doit(){
//        System.out.println("INCREASE POLL");
//        pollerMetadata().setMaxMessagesPerPoll(pollerMetadata().getMaxMessagesPerPoll() * 2);
//    }

    @Bean(name = PollerMetadata.DEFAULT_POLLER)
    public PollerMetadata pollerMetadata(){
        PollerMetadata metadata = new PollerMetadata();
        metadata.setTrigger(new DynamicPeriodicTrigger(Duration.ofSeconds(30)));
        metadata.setMaxMessagesPerPoll(1);
        return metadata;
    }

    @Bean
    public IntegrationFlow aggregatingFlow(
            AmqpInboundChannelAdapter aggregatorInboundChannel,
            AmqpOutboundEndpoint aggregatorOutboundChannel,
            MessageChannel wtChannel,
            MessageChannel aggregatingChannel,
            PollerMetadata pollerMetadata
    ) {
    return IntegrationFlows.from(aggregatorInboundChannel)
        .wireTap(wtChannel)
        .channel(aggregatingChannel)
        .handle(aggregatorOutboundChannel)
        .get();
    }

Answer 1

好的，如果有一条新的消息要组完成，它会到达聚合器，那么这样的组会立即被释放（如果您的ReleaseStrategy这样说的话）。处于超时状态的组的其余部分将继续等待计划。

可能会提出一种智能算法，利用MessageGroupStoreReaper依赖一个通用的时间表来决定我们是需要释放该部分组还是放弃它。再次强调：ReleaseStrategy应该给我们一个发布或不发布的线索，即使是部分发布。当丢弃发生时，我们希望将这些消息保留在聚合器中，我们需要在延迟一段时间后将它们重新发送回聚合器。过期后，组将从存储中删除，这发生在我们已经发送到丢弃通道时，因此最好延迟这些组并让聚合器清理这些组，这样在延迟后，我们可以安全地将它们作为新组的一部分发送回聚合器，以达到新的过期期。

您可能还可以在发布正常组后迭代存储区中的所有消息，以调整其标题中的某个时间键，以供下一个过期时间使用。

我知道这很难，但实际上没有现成的解决方案，因为它不是为了影响我们刚刚处理过的其他群体而设计的。。。