RabbitMQ骆驼消费者错误-java。网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当前在从队列中弹出一个值后进行一个简单的rest调用。它可以处理大约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"

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

因此，这将限制推送到使用者的消息数量，并防止缓冲区溢出。