3

这是我当前的 spring amqp 配置

<rabbit:connection-factory id="rabbitConnectionFactory"
    port="${rabbitmq.port}" host="${rabbitmq.host}" username="${rabbitmq.username}" password="${rabbitmq.password}"/>

<rabbit:admin id="rabbitmqAdmin" connection-factory="rabbitConnectionFactory" />

<rabbit:template id="importAmqpTemplate"
    connection-factory="rabbitConnectionFactory">
</rabbit:template>

这是我的交换、队列、侦听器、replyQueues、replyHandlers 配置

<rabbit:queue name="${process1.queue}" />
<rabbit:queue name="${process1.reply.queue}" />

<rabbit:queue name="${process2.queue}" />
<rabbit:queue name="${process2.reply.queue}" />

<rabbit:direct-exchange name="${myExchange}">
    <rabbit:bindings>
        <rabbit:binding queue="${process1.queue}"
            key="${process1.routing.key}" />
        <rabbit:binding queue="${process2.queue}"
            key="${process2.routing.key}" />
    </rabbit:bindings>
</rabbit:direct-exchange>

<rabbit:listener-container
    connection-factory="rabbitConnectionFactory" concurrency="${my.listener.concurrency}"
    requeue-rejected="false">
    <rabbit:listener queues="${process1.queue}"
        ref="foundation" method="process1" />
    <rabbit:listener queues="${process2.queue}"
        ref="foundation" method="process2s" />
</rabbit:listener-container>


<beans:beans profile="master">

    <beans:bean id="process1Lbq" class="java.util.concurrent.LinkedBlockingQueue" />
    <beans:bean id="process2Lbq" class="java.util.concurrent.LinkedBlockingQueue" />

    <beans:bean id="process1sReplyHandler"
        class="com.stockopedia.batch.foundation.ReplyHandler"
        p:blockingQueue-ref="process1Lbq" />

    <beans:bean id="process2ReplyHandler"
        class="com.stockopedia.batch.foundation.ReplyHandler"
        p:blockingQueue-ref="process2Lbq" />

    <rabbit:listener-container
        connection-factory="rabbitConnectionFactory" concurrency="1"
        requeue-rejected="false">
        <rabbit:listener queues="${process1.reply.queue}"
            ref="process1sHandler" method="onMessage" />
        <rabbit:listener queues="${process2.reply.queue}"
            ref="process2ReplyHandler" method="onMessage" />
    </rabbit:listener-container>

</beans:beans>

我已经在 6 个不同的服务器上进行了设置，并且只对来自主服务器的消息进行排队。其他服务器只处理消息。所有服务器都运行由并发设置的相同数量的侦听器。

问题是，消息需要不同的时间来处理。有些消息需要很长时间。因此，目前一些服务器不会从队列中获取消息，即使这些服务器上的所有侦听器都已完成处理那里的消息。

我可以看到待处理队列中的待处理消息，并且一些服务器处于空闲状态。我希望这些服务器在其他服务器忙于处理它们的消息时接收剩余的消息。

我是否需要按照教程http://www.rabbitmq.com/tutorials/tutorial-two-java.html (Fair Dispatch)中的说明设置 basic_Quos ？

int prefetchCount = 1;
channel.basicQos(prefetchCount);

还是 spring ampq 的默认设置？如果不是我该怎么做？

4

1 回答 1

1

basicQos(1)是侦听器容器的默认设置；可以通过prefetch容器上的设置来改变它。

我可以看到待处理队列中的待处理消息，并且一些服务器处于空闲状态。

如果您有空闲的消费者，您不应该只看到队列中的消息。如果消息被标记为未确认，则它们正在被处理。

如果您打开 DEBUG 级别的日志记录，您将能够看到空闲的消费者轮询内部队列以获取新的交付。

于 2014-05-19T12:42:27.243 回答