严格的消息排序

本节介绍入站和出站消息的消息排序。spring-doc.cn

入境

如果需要对入站消息进行严格排序，则必须将入站侦听器容器的属性配置为。这是因为，如果消息失败并重新传递，它将在现有的预取消息之后到达。从 Spring AMQP 版本 2.0 开始，默认为以提高性能。严格的订购要求以性能下降为代价。prefetchCount1prefetchCount250spring-doc.cn

出境

请考虑以下集成流程：spring-doc.cn

@Bean
public IntegrationFlow flow(RabbitTemplate template) {
    return IntegrationFlow.from(Gateway.class)
            .splitWith(s -> s.delimiters(","))
            .<String, String>transform(String::toUpperCase)
            .handle(Amqp.outboundAdapter(template).routingKey("rk"))
            .get();
}

假设我们发送消息，并且到网关。虽然消息、很可能是按顺序发送的，但不能保证。这是因为模板为每个 send 操作从缓存中“借用”一个通道，并且不能保证每条消息都使用相同的通道。一种解决方案是在拆分器之前启动事务，但事务在 RabbitMQ 中成本高昂，并且性能会降低数百倍。ABCABCspring-doc.cn

为了以更有效的方式解决这个问题，从版本 5.1 开始， Spring 集成提供了 which is a . 请参阅处理消息建议。在拆分器之前应用时，它确保所有下游操作都在同一通道上执行，并且可以选择等待收到所有已发送消息的发布者确认（如果连接工厂配置为确认）。以下示例演示如何使用：BoundRabbitChannelAdviceHandleMessageAdviceBoundRabbitChannelAdvicespring-doc.cn

@Bean
public IntegrationFlow flow(RabbitTemplate template) {
    return IntegrationFlow.from(Gateway.class)
            .splitWith(s -> s.delimiters(",")
                    .advice(new BoundRabbitChannelAdvice(template, Duration.ofSeconds(10))))
            .<String, String>transform(String::toUpperCase)
            .handle(Amqp.outboundAdapter(template).routingKey("rk"))
            .get();
}

请注意，在 advice 和 outbound adapter 中使用相同的（实现）。该建议在模板的方法中运行下游流，以便所有操作都在同一通道上运行。如果提供了可选的超时，则当流完成时，通知将调用该方法，如果在指定时间内未收到确认，该方法将引发异常。RabbitTemplateRabbitOperationsinvokewaitForConfirmsOrDiespring-doc.cn

下游流中不得有线程 hands-off（、和其他流）。QueueChannelExecutorChannel