Spring Amqp sendAndReceive操作获取null但没有发生超时

Posted 2023-02-16

【中文标题】Spring Amqp sendAndReceive操作获取null但没有发生超时

【英文标题】：Spring Amqp sendAndReceive operation get null but no timeout happens

【发布时间】：2015-02-20 13:07:36

【问题描述】：

正如 spring amqp 参考文档所说“当回复超时 (replyTimeout) 时，sendAndReceive() 方法返回 null”。 据我了解，replyTimeout 的默认值为 5 秒。

所以，我正在做的是打开 100 个线程。每个线程使用 sendAndReceive 操作将消息发送到一个队列。 RabbitListener 监听该队列，接收消息并发送回复。

并且一些线程在 5 秒之前就变为空。据我了解 RabbitTemplate 首先必须等待 5 秒，如果没有收到回复，则返回 null。请帮帮我，我做错了什么？

上下文：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xmlns:context="http://www.springframework.org/schema/context"
   xmlns:rabbit="http://www.springframework.org/schema/rabbit"

   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
                       http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
                       http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

<context:component-scan base-package="ru.cib" />
<rabbit:annotation-driven />
<rabbit:connection-factory id="connFactory" host="localhost" username="guest" password="guest" channel-cache-size="10" />
<rabbit:admin id="admin" connection-factory="connFactory" />    
<rabbit:queue name="requestQueue" />  
<rabbit:direct-exchange name="directExchange">
    <rabbit:bindings>
        <rabbit:binding key="directRequestKey" queue="requestQueue" />
    </rabbit:bindings>
</rabbit:direct-exchange>

<rabbit:template id="requestTemplate" connection-factory="connFactory" channel-transacted="true" reply-timeout="10000" />

<bean id="rabbitTxManager" class="org.springframework.amqp.rabbit.transaction.RabbitTransactionManager">
    <property name="connectionFactory" ref="connFactory" />
</bean>

<bean id="rabbitListenerContainerFactory" class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
        <property name="connectionFactory" ref="connFactory" />
        <property name="concurrentConsumers" value="10" />
        <property name="maxConcurrentConsumers" value="20" />
        <property name="transactionManager" ref="rabbitTxManager" />
        <property name="channelTransacted" value="true" />
    </bean>
</beans>

方法：

for (int i = 1; i <= 100; i++) 
    pool.submit(new RunnableImpl(i));


public void run() 
    LOG.info("thread send request ", i);
    Integer resp = (Integer) requestTemplate.convertSendAndReceive("directExchange", "directRequestKey", i);
    if (resp == null) 
        LOG.error("thread get null for request !!!", i);
     else 
        LOG.info("thread get response ", resp);
    


@RabbitListener(queues = "requestQueue")
    public int receivedRequestMessage(int data, Message request) throws Exception 
        LOG.info("receive request , ", data, request);
        LOG.info("sleep for 2 seconds");
        Thread.sleep(2000);
        int reply = data * 100;
        LOG.info("sending response ", reply);
        return reply;
    

【问题讨论】：

我猜日志中应该有一些“噪音”。我们有完全相同的测试用例：github.com/spring-projects/spring-amqp/blob/master/…。而且效果很好。

@ArtemBilan：非常感谢您的评论。我认为 sendAndReceive() 将像您提供的页面上第 120 行的测试用例一样（将在超时发生之前尝试多次接收答案）。我错了吗？

【参考方案1】：

我的错。 sendAndReceive 操作效果很好：等待 replyTimeout 属性中指定的时间，然后尝试接收答案。

我被自己的日志弄糊涂了……