Spring rabbitmq 消息排序不再起作用

Question

我正在处理一个消息排序问题，在不久前修复了它之后，现在修复不再起作用了。

只是为了概述，我有以下环境：

顺序在 tcpAdapter 和消息接收者之间的某处丢失。

这个问题我已经解决了：

1. 在生产者方面 - 使用发布者确认并返回

  rabbitmq:
    publisher-confirms: true
    publisher-returns: true

1. 在消费者端 - 强制执行单线程执行器： 我在这里找到的想法：RabbitMQ - Message order of delivery，为此我使用了后处理器。

@Component
public class RabbitConnectionFactoryPostProcessor implements BeanPostProcessor {
  @Override
  public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
    if (bean instanceof CachingConnectionFactory) {
      ((CachingConnectionFactory) bean).setExecutor(Executors.newSingleThreadExecutor());
    }
    return bean;
  }
}

现在，在一些 master-pom 更新后（我们不控制 master pom，它处于项目级别），修复突然不再起作用。检查差异后，我没有看到 spring-rabbit 或 spring-amqp 有任何变化，我不明白为什么会有影响。

---

如果你想要具体的例子，这里有更多细节：

1. 制片人。

TCP Server 向 tcpAdapter 应用程序发送消息，该应用程序使用 spring-integration 流从 TCP 获取消息并将其发送到 rabbitmq。

这是执行此操作的代码（inboundAdapterClient 我没有在此处发布，因为我认为它不重要）：

  @Bean
  public IntegrationFlow tcpToRabbitFlowClient() {
    return IntegrationFlows.from(inboundAdapterClient())      
        .transform(tcpToRabbitTransformer)     
        .channel(TCP_ADAPTER_SOURCE);
        .get();
  }

tcpAdapter 应用程序以正确的顺序从 TCP 接收消息，但是 tcpAdapter rabbitmq 堆栈每次都不会以正确的顺序发送它们（80% 的时间正常，20% 的错误顺序）

这里是 spring boot yml 配置（仅相关信息）：

spring:
  rabbitmq:
    publisher-confirms: true
    publisher-returns: true
  cloud:
    stream:
      bindings:
        tcpAdapterSource:
          binder: rabbit
          content-type: application/json
          destination: tcpadapter.messagereceiver

1. 消费者。

消息接收者强制执行单线程执行器以及如下配置。

这里是spring boot yml配置（只有相关信息）

spring:
  cloud:     
    stream:
      bindings:
        fromTcpAdapter:
          binder: rabbit
          content-type: application/json
          destination: tcpadapter.messagereceiver
      rabbit:
        default:
          producer:
            exchangeDurable: false
            exchangeAutoDelete: true
          consumer:
            exchangeDurable: false
            exchangeAutoDelete: true

注意：只有一个生产者和一个消费者。

来自 pom 的一些版本，也许有帮助：

      <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot</artifactId>
        <version>2.2.4.RELEASE</version>
      </dependency>
      <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-amqp</artifactId>
        <version>2.2.3.RELEASE</version>
      </dependency>
      <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>2.2.3.RELEASE</version>
      </dependency>
      <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream</artifactId>
        <version>3.0.1.RELEASE</version>
      </dependency>

Answer 1

通过删除 yml 配置并使用如下所述的显式 bean 声明和工厂配置来解决。唯一的问题是性能缓慢，但发布商确认这是预期的。

所以确实是生产者问题。

  @Bean
  public CachingConnectionFactory connectionFactory() {
    com.rabbitmq.client.ConnectionFactory connectionFactoryClient = new com.rabbitmq.client.ConnectionFactory();
    connectionFactoryClient.setUsername(username);
    connectionFactoryClient.setPassword(password);
    connectionFactoryClient.setHost(hostname);
    connectionFactoryClient.setVirtualHost(vhost);
    return new CachingConnectionFactory(connectionFactoryClient);
  }

  @Bean("rabbitTemplateAdapter")
  @Primary
  public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
    connectionFactory.setPublisherConfirmType(CORRELATED);
    connectionFactory.setPublisherReturns(true);
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMandatory(true);
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause)
            -> log.debug("correlationData({}),ack({}),cause ({})", correlationData, ack, cause));
    rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey)
            -> log.debug("exchange({}),route({}),replyCode({}),replyText({}),message:{}",
            exchange, routingKey, replyCode, replyText, message));
    return rabbitTemplate;
  }

对于发送消息：

rabbitTemplateAdapter.invoke(t -> {
      t.convertAndSend(
              exchange,
              DESTINATION,
              jsonMessage.getPayload(),
              m -> {outboundMapper().fromHeadersToRequest(jsonMessage.getHeaders(), m.getMessageProperties());
                return m;
              });
      t.waitForConfirmsOrDie(10_000);
      return true;
    });

我使用 spring rabbit 和 amqp 版本做到了这一点：

<dependency>
  <groupId>org.springframework.amqp</groupId>
  <artifactId>spring-rabbit</artifactId>
  <version>2.2.3.RELEASE</version>
</dependency>
<dependency>
  <groupId>org.springframework.amqp</groupId>
  <artifactId>spring-amqp</artifactId>
  <version>2.2.3.RELEASE</version>
</dependency>

Spring amqp 文档帮助很大，使用的技术称为“Scoped Operations”： https://docs.spring.io/spring-amqp/docs/2.2.7.RELEASE/reference/html/#scoped-operations