RabbitMQ 骆驼消费者错误 - java.net.SocketException：连接重置

Question

Camel RabbitMQ 的新手。使用 Apache Camel 编写了一个简单的 RabbitMQ 消费者。

onException(StateException.class).log(LoggingLevel.WARN,"WarnMessage = Error on ID and Status ${body}.")
            .asyncDelayedRedelivery().redeliveryDelay(5000).maximumRedeliveries(1)
            .setHeader(RabbitMQConstants.REQUEUE, constant(true))
            .handled(true)
            .setFaultBody(constant(true));

from(STATE__Q_URL).delay(200).log(LoggingLevel.DEBUG,"Incoming Body is --> ${body}")
          /*  .wireTap("direct:logtofile")*/.to("invokeManager")
            .log(LoggingLevel.INFO, "Message = Completed for ${body}");

invokeManager 当前在从队列中弹出一个值后进行简单的休息调用。 它适用于大约 100 条消息并开始抛出此错误

java.net.SocketException: Connection reset

java.net.SocketException: Broken pipe (Write failed)
    at java.net.SocketOutputStream.socketWrite0(Native Method) ~[na:1.8.0_144]
    at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:111) ~[na:1.8.0_144]
    at java.net.SocketOutputStream.write(SocketOutputStream.java:155) ~[na:1.8.0_144]
    at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82) ~[na:1.8.0_144]
    at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140) ~[na:1.8.0_144]
    at java.io.DataOutputStream.flush(DataOutputStream.java:123) ~[na:1.8.0_144]
    at com.rabbitmq.client.impl.SocketFrameHandler.flush(SocketFrameHandler.java:177) ~[amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.AMQConnection.flush(AMQConnection.java:559) ~[amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.AMQCommand.transmit(AMQCommand.java:127) ~[amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.AMQChannel.quiescingTransmit(AMQChannel.java:396) ~[amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.AMQChannel.quiescingTransmit(AMQChannel.java:378) ~[amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.AMQChannel.quiescingRpc(AMQChannel.java:313) ~[amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:601) ~[amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:534) ~[amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.StrictExceptionHandler.handleChannelKiller(StrictExceptionHandler.java:68) [amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.StrictExceptionHandler.handleConsumerException(StrictExceptionHandler.java:58) [amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:154) [amqp-client-4.2.1.jar!/:4.2.1]
    at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:100) [amqp-client-4.2.1.jar!/:4.2.1]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_144]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_144]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_144]

2018-02-25 18:06:17.551  WARN 4155 --- [0.42.31.42:5672] c.r.c.impl.ForgivingExceptionHandler     : An unexpected connection driver error occured (Exception message: Connection reset)

不知道我做错了什么，感谢任何帮助。

Answer 1

@Robbo_UK 是的，我发现问题出在消费者的缓冲区大小上。您需要设置一些字段才能解决此问题

            +"&prefetchEnabled=true"
            +"&prefetchCount=100"
            +"&threadPoolSize="100"
            +"&channelPoolMaxSize="100"

我从我的代码中随机复制了这些值。这将解决您的问题。

所以这将限制推送给消费者的消息数量，并防止缓冲区溢出。