本文介绍了骆驼中的无限循环-Rabbit MQ的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！
问题描述
我有一条小型服务器路由，该路由从queue.in接收消息并放入queue.out.
I have a small server-route, which takes a message from queue.in and puts in queue.out.
当我在 queue.in 中放入一条消息时，服务器将无限循环运行.我想知道我在配置方面缺少什么.这似乎是一条非常简单的路线.
When I put ONE message in queue.in, the server runns in an infinite loop. I wonder what I'm missing in terms of configuration. This seems like a very simple route.
小型服务器路由:
<camelContext id="camel-server" xmlns="http://camel.apache.org/schema/spring">
<jmxAgent disabled="false" onlyRegisterProcessorWithCustomId="false"
createConnector="true" usePlatformMBeanServer="true"
registerNewRoutes="true" statisticsLevel="All" includeHostName="false"
mask="false" id="agent" />
<endpoint id="queue.in" uri="${queue.url.input_queue}" />
<endpoint id="queue.out" uri="${queue.url.output_queue}" />
<route id="rabbitRoute">
<from uri="ref:queue.in" />
<bean ref="multiplier" /><!-- a simple bean that appends "BeanHello" to input msg-->
<to uri="ref:queue.out" />
</route>
</camelContext>
兔子队列属性:
queue.url.input_queue=rabbitmq://localhost:5672/ex1?queue=input_queue&routingKey=input_queue&connectionFactory=#customConnectionFactory&autoDelete=false&autoAck=true
queue.url.output_queue=rabbitmq://localhost:5672/ex1?queue=output_queue&routingKey=output_queue&connectionFactory=#customConnectionFactory&autoDelete=false&autoAck=true
日志文件的摘要-请参阅日志末尾的递归部分:字符串"BeanHello"被多次添加，这意味着消息将一次又一次地进入queue.in，而不是进入queue.out.
Snippet of log file - see recursive part at the end of log:The string "BeanHello" is added multiple times, that means the message is going to queue.in again and again instead of going to queue.out.
2014-12-14 15:00:16,692 [main ] DEBUG DefaultComponentResolver - Found component: rabbitmq via type: org.apache.camel.component.rabbitmq.RabbitMQComponent via: META-INF/services/org/apache/camel/component/rabbitmq
2014-12-14 15:00:16,699 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=components,name="rabbitmq"
2014-12-14 15:00:16,711 [main ] DEBUG DefaultComponent - Creating endpoint uri=[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue], path=[localhost:5672/ex1]
2014-12-14 15:00:16,721 [main ] DEBUG IntrospectionSupport - Configured property: autoAck on bean: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue] with value: true
2014-12-14 15:00:16,725 [main ] DEBUG IntrospectionSupport - Configured property: autoDelete on bean: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue] with value: false
2014-12-14 15:00:16,726 [main ] DEBUG IntrospectionSupport - Configured property: queue on bean: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue] with value: input_queue
2014-12-14 15:00:16,727 [main ] DEBUG IntrospectionSupport - Configured property: routingKey on bean: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue] with value: input_queue
2014-12-14 15:00:16,728 [main ] DEBUG RabbitMQComponent - Creating RabbitMQEndpoint with host localhost:5672 and exchangeName: ex1
2014-12-14 15:00:16,729 [main ] DEBUG SpringCamelContext - rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue converted to endpoint: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue] by component: org.apache.camel.component.rabbitmq.RabbitMQComponent@2be25129
2014-12-14 15:00:16,730 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=endpoints,name="rabbitmq://localhost:5672/ex1\?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue"
2014-12-14 15:00:16,730 [main ] DEBUG DefaultManagementAgent - MBean already registered with ObjectName: org.apache.camel:context=camel-server,type=endpoints,name="rabbitmq://localhost:5672/ex1\?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue"
2014-12-14 15:00:16,743 [main ] DEBUG SpringCamelContext - Using ComponentResolver: org.apache.camel.impl.DefaultComponentResolver@1bd7643a to resolve component with name: bean
2014-12-14 15:00:16,743 [main ] DEBUG DefaultComponentResolver - Found component: bean in registry: null
2014-12-14 15:00:16,743 [main ] DEBUG DefaultComponentResolver - Found component: bean via type: org.apache.camel.component.bean.BeanComponent via: META-INF/services/org/apache/camel/component/bean
2014-12-14 15:00:16,748 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=components,name="bean"
2014-12-14 15:00:16,758 [main ] DEBUG DefaultChannel - Initialize channel for target: 'Bean[ref:multiplier]'
2014-12-14 15:00:16,765 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=tracer,name=BacklogTracer
2014-12-14 15:00:16,774 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=tracer,name=BacklogDebugger
2014-12-14 15:00:16,788 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=errorhandlers,name="DefaultErrorHandlerBuilder(ref:CamelDefaultErrorHandlerBuilder)"
2014-12-14 15:00:16,789 [main ] DEBUG DefaultComponent - Creating endpoint uri=[ref://queue.out], path=[queue.out]
2014-12-14 15:00:16,789 [main ] DEBUG SpringCamelContext - ref://queue.out converted to endpoint: Endpoint[ref://queue.out] by component: org.apache.camel.component.ref.RefComponent@2182fb16
2014-12-14 15:00:16,790 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=endpoints,name="ref://queue.out"
2014-12-14 15:00:16,792 [main ] DEBUG DefaultComponent - Creating endpoint uri=[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue], path=[localhost:5672/ex1]
2014-12-14 15:00:16,794 [main ] DEBUG IntrospectionSupport - Configured property: autoAck on bean: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue] with value: true
2014-12-14 15:00:16,795 [main ] DEBUG IntrospectionSupport - Configured property: autoDelete on bean: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue] with value: false
2014-12-14 15:00:16,797 [main ] DEBUG IntrospectionSupport - Configured property: queue on bean: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue] with value: output_queue
2014-12-14 15:00:16,798 [main ] DEBUG IntrospectionSupport - Configured property: routingKey on bean: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue] with value: output_queue
2014-12-14 15:00:16,798 [main ] DEBUG RabbitMQComponent - Creating RabbitMQEndpoint with host localhost:5672 and exchangeName: ex1
2014-12-14 15:00:16,800 [main ] DEBUG SpringCamelContext - rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue converted to endpoint: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue] by component: org.apache.camel.component.rabbitmq.RabbitMQComponent@2be25129
2014-12-14 15:00:16,801 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=endpoints,name="rabbitmq://localhost:5672/ex1\?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue"
2014-12-14 15:00:16,801 [main ] DEBUG DefaultManagementAgent - MBean already registered with ObjectName: org.apache.camel:context=camel-server,type=endpoints,name="rabbitmq://localhost:5672/ex1\?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue"
2014-12-14 15:00:16,802 [main ] DEBUG DefaultChannel - Initialize channel for target: 'To[ref:queue.out]'
2014-12-14 15:00:16,808 [main ] INFO SpringCamelContext - AllowUseOriginalMessage is enabled. If access to the original message is not needed, then its recommended to turn this option off as it may improve performance.
2014-12-14 15:00:16,808 [main ] INFO SpringCamelContext - StreamCaching is not in use. If using streams then its recommended to enable stream caching. See more details at http://camel.apache.org/stream-caching.html
2014-12-14 15:00:16,809 [main ] DEBUG SpringCamelContext - Warming up route id: rabbitRoute having autoStartup=true
2014-12-14 15:00:16,809 [main ] DEBUG RouteService - Starting services on route: rabbitRoute
2014-12-14 15:00:16,811 [main ] DEBUG RouteService - Starting child service on route: rabbitRoute -> Pipeline[[Channel[BeanProcessor[org.apache.camel.example.server.Treble(0x64eb29e1)]], Channel[sendTo(Endpoint[ref://queue.out])]]]
2014-12-14 15:00:16,811 [main ] DEBUG DefaultErrorHandler - Redelivery enabled: false on error handler: DefaultErrorHandler[Instrumentation:bean[BeanProcessor[org.apache.camel.example.server.Treble(0x64eb29e1)]]]
2014-12-14 15:00:16,818 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=producers,name=RabbitMQProducer(0x89c08d3)
2014-12-14 15:00:16,823 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=threadpools,name="RabbitMQProducer(0x89c08d3)"
2014-12-14 15:00:16,824 [main ] DEBUG DefaultExecutorServiceManager - Created new ThreadPool for source: Producer[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue] with name: CamelRabbitMQProducer[output_queue]. -> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@69e4d7d[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0][CamelRabbitMQProducer[output_queue]]
2014-12-14 15:00:16,890 [main ] DEBUG RabbitMQProducer - Created connection: amqp://[email protected]:5672/
2014-12-14 15:00:16,911 [main ] DEBUG ProducerCache - Adding to producer cache with key: Endpoint[ref://queue.out] for producer: Producer[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=output_queue&routingKey=output_queue]
2014-12-14 15:00:16,912 [main ] DEBUG DefaultErrorHandler - Redelivery enabled: false on error handler: DefaultErrorHandler[Instrumentation:to[sendTo(Endpoint[ref://queue.out])]]
2014-12-14 15:00:16,912 [main ] DEBUG RouteService - Starting child service on route: rabbitRoute -> Pipeline[[Channel[BeanProcessor[org.apache.camel.example.server.Treble(0x64eb29e1)]], Channel[sendTo(Endpoint[ref://queue.out])]]]
2014-12-14 15:00:16,912 [main ] DEBUG RouteService - Starting child service on route: rabbitRoute -> BeanProcessor[org.apache.camel.example.server.Treble(0x64eb29e1)]
2014-12-14 15:00:16,919 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=processors,name="bean1"
2014-12-14 15:00:16,920 [main ] DEBUG RouteService - Starting child service on route: rabbitRoute -> Channel[BeanProcessor[org.apache.camel.example.server.Treble(0x64eb29e1)]]
2014-12-14 15:00:16,920 [main ] DEBUG RouteService - Starting child service on route: rabbitRoute -> sendTo(Endpoint[ref://queue.out])
2014-12-14 15:00:16,928 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=processors,name="to1"
2014-12-14 15:00:16,928 [main ] DEBUG RouteService - Starting child service on route: rabbitRoute -> Channel[sendTo(Endpoint[ref://queue.out])]
2014-12-14 15:00:16,939 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=routes,name="rabbitRoute"
2014-12-14 15:00:16,939 [main ] DEBUG TimerListenerManager - Added TimerListener: org.apache.camel.management.mbean.ManagedSuspendableRoute@19f7f0f0
2014-12-14 15:00:16,940 [main ] DEBUG SpringCamelContext - Route: rabbitRoute >>> EventDrivenConsumerRoute[Endpoint[ref://queue.in] -> Pipeline[[Channel[BeanProcessor[org.apache.camel.example.server.Treble(0x64eb29e1)]], Channel[sendTo(Endpoint[ref://queue.out])]]]]
2014-12-14 15:00:16,940 [main ] DEBUG SpringCamelContext - Starting consumer (order: 1000) on route: rabbitRoute
2014-12-14 15:00:16,945 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=consumers,name=RabbitMQConsumer(0x753af01f)
2014-12-14 15:00:16,947 [main ] DEBUG DefaultManagementAgent - Registered MBean with ObjectName: org.apache.camel:context=camel-server,type=threadpools,name="RabbitMQEndpoint(0xad9be15)"
2014-12-14 15:00:16,947 [main ] DEBUG DefaultExecutorServiceManager - Created new ThreadPool for source: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue] with name: RabbitMQConsumer. -> org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@5d4e5a43[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0][RabbitMQConsumer]
2014-12-14 15:00:16,947 [main ] DEBUG RabbitMQConsumer - Using executor org.apache.camel.util.concurrent.RejectableThreadPoolExecutor@5d4e5a43[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0][RabbitMQConsumer]
2014-12-14 15:00:16,951 [main ] DEBUG RabbitMQConsumer - Created connection: amqp://[email protected]:5672/
2014-12-14 15:00:16,952 [main ] DEBUG RabbitMQConsumer - Created channel: AMQChannel(amqp://[email protected]:5672/,1)
2014-12-14 15:00:16,957 [main ] INFO SpringCamelContext - Route: rabbitRoute started and consuming from: Endpoint[rabbitmq://localhost:5672/ex1?autoAck=true&autoDelete=false&connectionFactory=%23customConnectionFactory&queue=input_queue&routingKey=input_queue]
2014-12-14 15:00:16,958 [main ] DEBUG ultManagementLifecycleStrategy - Load performance statistics disabled
2014-12-14 15:00:16,958 [main ] INFO SpringCamelContext - Total 1 routes, of which 1 is started.
2014-12-14 15:00:16,962 [main ] INFO SpringCamelContext - Apache Camel 2.15-SNAPSHOT (CamelContext: camel-server) started in 0.889 seconds
2014-12-14 15:00:16,963 [main ] DEBUG MainSupport - Starting Spring ApplicationContext: org.springframework.context.support.ClassPathXmlApplicationContext@3db853a5
2014-12-14 15:00:16,964 [main ] DEBUG SpringCamelContext - onApplicationEvent: org.springframework.context.event.ContextStartedEvent[source=org.springframework.context.support.ClassPathXmlApplicationContext@3db853a5: startup date [Sun Dec 14 15:00:14 PST 2014]; root of context hierarchy]
2014-12-14 15:01:01,671 [abbitMQConsumer] DEBUG MethodInfo - Setting bean invocation result on the IN message: @@@MsgFromClient#BeanHello
2014-12-14 15:01:01,676 [abbitMQConsumer] DEBUG SendProcessor - >>>> Endpoint[ref://queue.out] Exchange[Message: @@@MsgFromClient#BeanHello ]
2014-12-14 15:01:01,683 [abbitMQConsumer] DEBUG MethodInfo - Setting bean invocation result on the IN message: @@@MsgFromClient#BeanHello #BeanHello
2014-12-14 15:01:01,684 [abbitMQConsumer] DEBUG SendProcessor - >>>> Endpoint[ref://queue.out] Exchange[Message: @@@MsgFromClient#BeanHello #BeanHello ]
2014-12-14 15:01:01,687 [abbitMQConsumer] DEBUG MethodInfo - Setting bean invocation result on the IN message: @@@MsgFromClient#BeanHello #BeanHello #BeanHello
2014-12-14 15:01:01,687 [abbitMQConsumer] DEBUG SendProcessor - >>>> Endpoint[ref://queue.out] Exchange[Message: @@@MsgFromClient#BeanHello #BeanHello #BeanHello ]
2014-12-14 15:01:01,698 [abbitMQConsumer] DEBUG MethodInfo - Setting bean invocation result on the IN message: @@@MsgFromClient#BeanHello #BeanHello #BeanHello #BeanHello
2014-12-14 15:01:01,698 [abbitMQConsumer] DEBUG SendProcessor - >>>> Endpoint[ref://queue.out] Exchange[Message: @@@MsgFromClient#BeanHello #BeanHello #BeanHello #BeanHello ]
推荐答案
解决方案:
更改了uri，添加了BridgedEndpoint = true.我不了解其背后的机制，也不了解为什么 rabbitmq会带来额外的并发症.如果我使用jboss-fuse/camel，我应该期待同样的问题吗?
Changed the uri added BridgedEndpoint=true. I do not understand the mechanics behind that, and why such extra complications for rabbitmq specifically. If I use jboss-fuse/camel I should expect the same issue?
queue.url.input_queue=rabbitmq://localhost:5672/ex1?queue=input_queue&routingKey=input_queue&connectionFactory=#customConnectionFactory&autoDelete=false&autoAck=true&BridgeEndpoint=true
queue.url.output_queue=rabbitmq://localhost:5672/ex1?queue=output_queue&routingKey=output_queue&connectionFactory=#customConnectionFactory&autoDelete=false&autoAck=true&BridgeEndpoint=true
这篇关于骆驼中的无限循环-Rabbit MQ的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！