java - AggregatingMessageHandler的手动ACK

我正在尝试构建这样的集成方案，例如Rabbit -> AmqpInboundChannelAdapter(AcknowledgeMode.MANUAL) -> DirectChannel -> AggregatingMessageHandler -> DirectChannel -> AmqpOutboundEndpoint。

我想聚合内存中的消息并在我聚合10条消息或达到10秒超时时释放它。我想这个配置是可以的：

@Bean
@ServiceActivator(inputChannel = "amqpInputChannel")
public MessageHandler aggregator(){
    AggregatingMessageHandler aggregatingMessageHandler = new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor(), new SimpleMessageStore(10));
    aggregatingMessageHandler.setCorrelationStrategy(new HeaderAttributeCorrelationStrategy(AmqpHeaders.CORRELATION_ID));
    //default false
    aggregatingMessageHandler.setExpireGroupsUponCompletion(true);  //when grp released (using strategy), remove group so new messages in same grp create new group
    aggregatingMessageHandler.setSendPartialResultOnExpiry(true);   //when expired because timeout and not because of strategy, still send messages grouped so far
    aggregatingMessageHandler.setGroupTimeoutExpression(new ValueExpression<>(TimeUnit.SECONDS.toMillis(10)));  //timeout after X

    //timeout is checked only when new message arrives!!
    aggregatingMessageHandler.setReleaseStrategy(new TimeoutCountSequenceSizeReleaseStrategy(10, TimeUnit.SECONDS.toMillis(10)));
    aggregatingMessageHandler.setOutputChannel(amqpOutputChannel());
    return aggregatingMessageHandler;
}

现在，我的问题是-除了以这种方式创建自己的实现AggregatingMessageHandler的实现之外，还有什么更简便的方法来手动确认消息：

public class ManualAckAggregatingMessageHandler extends AbstractCorrelatingMessageHandler {
        ...

    private void ackMessage(Channel channel, Long deliveryTag){
        try {
            Assert.notNull(channel, "Channel must be provided");
            Assert.notNull(deliveryTag, "Delivery tag must be provided");
            channel.basicAck(deliveryTag, false);
        }
        catch (IOException e) {
            throw new MessagingException("Cannot ACK message", e);
        }
    }

    @Override
    protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages) {
        Object groupId = messageGroup.getGroupId();
        MessageGroupStore messageStore = getMessageStore();
        messageStore.completeGroup(groupId);

        messageGroup.getMessages().forEach(m -> {
            Channel channel = (Channel)m.getHeaders().get(AmqpHeaders.CHANNEL);
            Long deliveryTag = (Long)m.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
            ackMessage(channel, deliveryTag);
        });

        if (this.expireGroupsUponCompletion) {
            remove(messageGroup);
        }
        else {
            if (messageStore instanceof SimpleMessageStore) {
                ((SimpleMessageStore) messageStore).clearMessageGroup(groupId);
            }
            else {
                messageStore.removeMessagesFromGroup(groupId, messageGroup.getMessages());
            }
        }
    }
}

更新

在您的帮助下，我设法做到了。最重要的部分：连接工厂必须具有factory.setPublisherConfirms(true)。 AmqpOutboundEndpoint必须具有以下两个设置：outboundEndpoint.setConfirmAckChannel(manualAckChannel())和outboundEndpoint.setConfirmCorrelationExpressionString("#root")，这是其余类的实现：

public class ManualAckPair {
    private Channel channel;
    private Long deliveryTag;

    public ManualAckPair(Channel channel, Long deliveryTag) {
        this.channel = channel;
        this.deliveryTag = deliveryTag;
    }

    public void basicAck(){
        try {
            this.channel.basicAck(this.deliveryTag, false);
        }
        catch (IOException e) {
            e.printStackTrace();
        }
    }
}

public abstract class AbstractManualAckAggregatingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor {
    public static final String MANUAL_ACK_PAIRS = PREFIX + "manualAckPairs";

    @Override
    protected Map<String, Object> aggregateHeaders(MessageGroup group) {
        Map<String, Object> aggregatedHeaders = super.aggregateHeaders(group);
        List<ManualAckPair> manualAckPairs = new ArrayList<>();
        group.getMessages().forEach(m -> {
            Channel channel = (Channel)m.getHeaders().get(AmqpHeaders.CHANNEL);
            Long deliveryTag = (Long)m.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
            manualAckPairs.add(new ManualAckPair(channel, deliveryTag));
        });
        aggregatedHeaders.put(MANUAL_ACK_PAIRS, manualAckPairs);
        return aggregatedHeaders;
    }
}

和

@Service
public class ManualAckServiceActivator {

    @ServiceActivator(inputChannel = "manualAckChannel")
public void handle(@Header(MANUAL_ACK_PAIRS) List<ManualAckPair> manualAckPairs) {
    manualAckPairs.forEach(manualAckPair -> {
        manualAckPair.basicAck();
    });
}
}

最佳答案

是的，聚合器不需要如此复杂的逻辑。

您只需在聚合器发布后就可以确认它们-在聚合器与该AmqpOutboundEndpoint之间的服务激活器中。

正确的是，您必须在其中使用basicAck()和multiple标志设置为true：

@param multiple true to acknowledge all messages up to and

好吧，为此，您肯定需要自定义MessageGroupProcessor来提取整个批次的最高AmqpHeaders.DELIVERY_TAG并将其设置为输出聚合消息的标头。

您可以扩展DefaultAggregatingMessageGroupProcessor并覆盖其aggregateHeaders()：

/**
 * This default implementation simply returns all headers that have no conflicts among the group. An absent header
 * on one or more Messages within the group is not considered a conflict. Subclasses may override this method with
 * more advanced conflict-resolution strategies if necessary.
 *
 * @param group The message group.
 * @return The aggregated headers.
 */
protected Map<String, Object> aggregateHeaders(MessageGroup group) {