在AsyncRabbitTemboard中回复消息后处理

Question

在使用AsyncRabbitTemplate时，我在GZip/GUnzip消息处理方面遇到了问题。

使用同步模板设置时一切正常，如下所示：

@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
    final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMessageConverter(jsonConverter());
    rabbitTemplate.setReplyTimeout(config.getRabbitSendAndReceiveReplyTimeout());
    rabbitTemplate.setReceiveTimeout(config.getRabbitSendAndReceiveReceiveTimeout());
    rabbitTemplate.setAfterReceivePostProcessors(new GUnzipPostProcessor(true));
    rabbitTemplate.setBeforePublishPostProcessors(new GZipPostProcessor(true));
    return rabbitTemplate;
}

但是，当我设置像这样的异步模板时：

@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public AsyncRabbitTemplate rabbitTemplateAsync(final ConnectionFactory connectionFactory) {
    final AsyncRabbitTemplate asyncRabbitTemplate = new AsyncRabbitTemplate(rabbitTemplate(connectionFactory));
    // need to manually start the reply listener container for some reason
    asyncRabbitTemplate.start();
    return asyncRabbitTemplate;
}

回复消息没有正确解压缩，我得到了这个错误消息

Caused by: java.io.UnsupportedEncodingException: gzip:UTF-8
    at java.lang.StringCoding.decode(Unknown Source) ~[?:1.8.0_192]
    at java.lang.String.<init>(Unknown Source) ~[?:1.8.0_192]
    at java.lang.String.<init>(Unknown Source) ~[?:1.8.0_192]
    at org.springframework.amqp.support.converter.AbstractJackson2MessageConverter.convertBytesToObject(AbstractJackson2MessageConverter.java:235) ~[spring-amqp-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.support.converter.AbstractJackson2MessageConverter.fromMessage(AbstractJackson2MessageConverter.java:199) ~[spring-amqp-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.AsyncRabbitTemplate.onMessage(AsyncRabbitTemplate.java:576) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]

我已尝试为AsyncRabbitTemplate提供一个配置好的DirectReplyToMessageListenerContainer，但没有帮助

    final DirectReplyToMessageListenerContainer directReplyToMessageListenerContainer = new DirectReplyToMessageListenerContainer(
            connectionFactory);
    directReplyToMessageListenerContainer.setAfterReceivePostProcessors(new GUnzipPostProcessor(true));
    final AsyncRabbitTemplate asyncRabbitTemplate = new AsyncRabbitTemplate(rabbitTemplate(connectionFactory),
            directReplyToMessageListenerContainer);

这只会导致以下错误：

[错误]2019-03-06 12:18:05.192[AMQP连接172.17.3.6:5672]CachingConnectionFactory.log-通道关闭：通道错误；协议方法：#方法（回复代码=406，回复文本=PRECONDITION_FAILED-快速回复消费者不存在，类id=60，方法id=40）

请注意，我通过采用SpringRabbit项目的一个分支，并将此构造函数添加到AsyncRabbitTemplate中，使事情得以顺利进行：

public IndigoAsyncRabbitTemplate(final RabbitTemplate template,
        final DirectReplyToMessageListenerContainer directReplyToContainer) {
    Assert.notNull(template, "'template' cannot be null");
    this.template = template;
    container = null;
    replyAddress = null;
    this.directReplyToContainer = directReplyToContainer;
    directReplyToContainer.setMessageListener(this);
}

那么，这是否需要对SpringRabbit库进行增强才能开始工作呢？或者，有没有一种方法可以让GUnzip在回复侦听器上工作，而不必跳过太多的障碍？

Answer 1

对，这必须作为框架的改进。在AsyncRabbitTemboard的情况下，我们只是忽略了ReceivePost处理器之后的这一事实。我们可以重新配置一个内部的DirectReplyToMessageListenerContainer，以便从提供的RabbitTemboard中使用在ReceivePost处理器之后。

同时，您可以坚持使用常规的SimpleMessageListenerContainer注入。或者您可以尝试使用外部DirectReplyToMessageListenerContainer注入。

见本节：

/**
 * Construct an instance using the provided arguments. The first queue the container
 * is configured to listen to will be used as the reply queue. Replies will be
 * routed using the default exchange with that queue name as the routing key.
 * @param template a {@link RabbitTemplate}
 * @param container a {@link AbstractMessageListenerContainer}.
 */
public AsyncRabbitTemplate(RabbitTemplate template, AbstractMessageListenerContainer container) {
    this(template, container, null);
}

关于这一事项的问题：https://github.com/spring-projects/spring-amqp/issues/920