首页 文章

Spring-amqp - 消息处理延迟

提问于
浏览
1

我们在RHEL 7.0 VM上部署了一个Java / spring / tomcat应用程序,该应用程序使用AlejandroRivera / embedded-rabbitmq,并在部署战争后立即启动Rabbitmq服务器,并连接到它 . 我们有多个队列用于处理和过滤事件 .

流程是这样的:

我们收到的事件 - >发布事件队列 - >监听器类过滤事件 - >发布到另一个队列进行处理 - >我们发布到另一个队列进行日志记录 .

问题是:

  • 处理正常开始,我们可以看到消息流过队列,但是经过一段时间后,侦听器类停止接收事件 . 看起来我们能够将它发布到RabbitMQ Channels ,但它永远不会从侦听器中排出队列 . 这似乎开始降级导致事件在一段时间后被处理,上升到几分钟 . 负载不是大约200个事件,我们关心它只有少数事件 .

我们尝试了什么:

  • 最初队列有 pre-fetch set to 1 ,消费者是 min of 2 and max of 5 ,我们删除了预取,我们添加了更多的消费者作为最大并发设置,但问题仍然存在,延迟只需要更长时间才能呈现,但几分钟后,处理开始需要大约20/30秒 .

我们在日志中看到我们将事件发布到队列中,并且我们看到了日志,我们将它从队列中延迟了 . 所以我们的代码中间没有任何东西在运行这个延迟 .

据我们所知,其余的队列似乎正确地处理了消息,但这是一个陷入这种卡住模式的队列 .

我看到的错误如下,但我知道它意味着什么以及它是否相关:

Jun  4 11:16:04  server: [pool-3-thread-10] ERROR com.rabbitmq.client.impl.ForgivingExceptionHandler - Consumer org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$InternalConsumer@70dfa413 (amq.ctag-VaWc-hv-VwcUPh9mTQTj7A) method handleDelivery for channel AMQChannel(amqp://agent@127.0.0.1:5672/,198) threw an exception for channel AMQChannel(amqp://agent@127.0.0.1:5672/,198)
Jun  4 11:16:04  server: java.io.IOException: Unknown consumerTag
Jun  4 11:16:04  server: at com.rabbitmq.client.impl.ChannelN.basicCancel(ChannelN.java:1266)
Jun  4 11:16:04  server: at sun.reflect.GeneratedMethodAccessor180.invoke(Unknown Source)
Jun  4 11:16:04  server: at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
Jun  4 11:16:04  server: at java.lang.reflect.Method.invoke(Method.java:498)
Jun  4 11:16:04  server: at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$CachedChannelInvocationHandler.invoke(CachingConnectionFactory.java:955)
Jun  4 11:16:04  server: at com.sun.proxy.$Proxy119.basicCancel(Unknown Source)
Jun  4 11:16:04  server: at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$InternalConsumer.handleDelivery(BlockingQueueConsumer.java:846)
Jun  4 11:16:04  server: at com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149)
Jun  4 11:16:04  server: at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:100)
Jun  4 11:16:04  server: at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
Jun  4 11:16:04  server: at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
Jun  4 11:16:04  server: at java.lang.Thread.run(Thread.java:748)

这个在关闭应用程序时发生,但我已经看到它在应用程序仍在运行时发生..

2018-06-05 13:22:45,443 ERROR CachingConnectionFactory$DefaultChannelCloseLogger - Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 109, class-id=60, method-id=120)

我不确定如何解决这两个错误,也不确定它们是否相关 .

这是我的Spring配置:

<!-- Queues -->
<rabbit:queue id="monitorIncomingEventsQueue" name="MonitorIncomingEventsQueue"/>
<rabbit:queue id="interestingEventsQueue" name="InterestingEventsQueue"/>
<rabbit:queue id="textCallsEventsQueue" name="TextCallsEventsQueue"/>
<rabbit:queue id="callDisconnectedEventQueue" name="CallDisconnectedEventQueue"/>
<rabbit:queue id="incomingCallEventQueue" name="IncomingCallEventQueue"/>
<rabbit:queue id="eventLoggingQueue" name="EventLoggingQueue"/>

<!-- listeners -->
<bean id="monitorListener" class="com.example.rabbitmq.listeners.monitorListener"/>
<bean id="interestingEventsListener" class="com.example.rabbitmq.listeners.InterestingEventsListener"/>
<bean id="textCallsEventListener" class="com.example.rabbitmq.listeners.TextCallsEventListener"/>
<bean id="callDisconnectedEventListener" class="com.example.rabbitmq.listeners.CallDisconnectedEventListener"/>
<bean id="incomingCallEventListener" class="com.example.rabbitmq.listeners.IncomingCallEventListener"/>
<bean id="eventLoggingEventListener" class="com.example.rabbitmq.listeners.EventLoggingListener"/>

<rabbit:listener-container connection-factory="connectionFactory" message-converter="defaultMessageConverter" concurrency="5" max-concurrency="40" acknowledge="none">
    <rabbit:listener queues="interestingEventsQueue" ref="interestingEventsListener" method="handleIncomingMessage"/>
</rabbit:listener-container>

<rabbit:listener-container connection-factory="connectionFactory" message-converter="defaultMessageConverter" concurrency="5" max-concurrency="20" acknowledge="none">
    <rabbit:listener queues="textCallsEventsQueue" ref="textCallsEventListener" method="handleIncomingMessage"/>
</rabbit:listener-container>

<rabbit:listener-container connection-factory="connectionFactory" message-converter="defaultMessageConverter" concurrency="5" max-concurrency="20" acknowledge="none">
    <rabbit:listener queues="callDisconnectedEventQueue" ref="callDisconnectedEventListener" method="handleIncomingMessage"/>
</rabbit:listener-container>

<rabbit:listener-container connection-factory="connectionFactory" message-converter="defaultMessageConverter" concurrency="5" max-concurrency="30" acknowledge="none">
    <rabbit:listener queues="incomingCallEventQueue" ref="incomingCallEventListener" method="handleIncomingMessage"/>
</rabbit:listener-container>

<rabbit:listener-container connection-factory="connectionFactory" message-converter="defaultMessageConverter" concurrency="1" max-concurrency="3" acknowledge="none">
    <rabbit:listener queues="monitorIncomingEventsQueue" ref="monitorListener" method="handleIncomingMessage"/>
</rabbit:listener-container>

<rabbit:listener-container connection-factory="connectionFactory" message-converter="defaultMessageConverter" concurrency="5" max-concurrency="10"  acknowledge="none">
    <rabbit:listener queues="EventLoggingQueue" ref="eventLoggingEventListener" method="handleLoggingEvent"/>
</rabbit:listener-container>

<rabbit:connection-factory id="connectionFactory" host="${host.name}" port="${port.number}" username="${user.name}" password="${user.password}" connection-timeout="20000"/>

我在这里读到,处理延迟可能是由网络问题引起的,但在这种情况下,服务器和应用程序位于同一个VM上 . 这是一个被锁定的环境,因此大多数端口都没有打开,但我怀疑这是错的 .

更多日志:https://pastebin.com/4QMFDT7A

任何帮助表示赞赏,

谢谢,

1 回答

  • 0

    我需要看到更多的日志 - 这是吸烟枪:

    Storing...Storing delivery for Consumer@a2ce092: tags=[{}]
    

    (使用者) tags 为空,这意味着消费者当时已被取消(出于某种原因,应该在日志中出现) .

    如果你有可能用1.7.9.BUILD-SNAPSHOT重现,我添加了一些TRACE级别的日志记录,这应该有助于诊断它 .

    EDIT

    回复你最近对rabbitmq-users的评论......

    你能尝试使用固定并发吗? Spring AMQP容器中的可变并发性通常不是很有用,因为如果整个容器空闲一段时间,通常只会减少消费者 .

    但是,它可能会解释为什么您会看到消费者被取消 .

    也许这种逻辑中存在一些竞争条件;使用固定数量的消费者(不指定max ...)将避免这种情况;如果你可以尝试,它至少会消除这种可能性 .

    也就是说,我很困惑(我在Stack Overflow配置中没有注意到这一点);使用 acknowledge="none" 应该没有acks被发送到代理(NONE用于设置 autoAck

    String consumerTag = this.channel.basicConsume(queue, this.acknowledgeMode.isAutoAck(), ...
    

    public boolean isAutoAck() {
    
        return this == NONE;
    
    }
    

    你是从你的代码发送ack?如果是这样,ack模式应该是MANUAL . 我看不到容器将发送确认为无通道模式的情况 .

相关问题