从 AMQP 到 RESTful 端点的骆驼路由
Camel Routing from AMQP to RESTful endpoint
我是 Camel 的新手，到目前为止一直玩得很开心。但是，我现在在尝试从 AMQP 读取消息并使用消息中的参数调用 REST 端点时遇到问题。
我的想法是 AMQP 消息包含一个 POJO，我正在阅读它以获得邮政编码和门牌号值。这些值然后用于填充 URL 中的动态字段。
我正在使用 HTTP4 执行对 REST 端点的调用，每次我点击该行时，我都会收到：
Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor Elapsed (ms)
[route1 ] [route1 ] [spring-amqp://myExchange:getAddress:camel.getAddress?autodelete=false&durable=] [ 46]
[route1 ] [log1 ] [log ] [ 13]
[route1 ] [setHeader1 ] [setHeader[CamelHttpUrl] ] [ 0]
[route1 ] [setExchangePattern] [setExchangePattern[InOut] ] [ 0]
[route1 ] [setHeader2 ] [setHeader[postcode] ] [ 18]
[route1 ] [setHeader3 ] [setHeader[houseNumber] ] [ 1]
[route1 ] [log2 ] [log ] [ 0]
[route1 ] [recipientList1 ] [recipientList[simple{Simple: https4:{{addressmicroservice.address.url}}/${head] [ 13]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.InvalidPayloadException: No body available of type: java.io.InputStream but has value: Address [houseNumber=null, postcode=DY10 4TW, line1=null, line2=null, line3=null, line4=null, locality=null, town=null, county=null] of type: uk.gov.insolvency.microservices.pojos.Address on: Message[ID-Andrews-MacBook-Pro-local-56293-1472721966485-0-3]. Caused by: No type converter available to convert from type: uk.gov.insolvency.microservices.pojos.Address to the required type: java.io.InputStream with value Address [houseNumber=null, postcode=DY10 4TW, line1=null, line2=null, line3=null, line4=null, locality=null, town=null, county=null]. Exchange[ID-Andrews-MacBook-Pro-local-56293-1472721966485-0-4]. Caused by: [org.apache.camel.NoTypeConversionAvailableException - No type converter available to convert from type: uk.gov.insolvency.microservices.pojos.Address to the required type: java.io.InputStream with value Address [houseNumber=null, postcode=DY10 4TW, line1=null, line2=null, line3=null, line4=null, locality=null, town=null, county=null]]
at org.apache.camel.impl.MessageSupport.getMandatoryBody(MessageSupport.java:107)
at org.apache.camel.component.http4.HttpProducer.createRequestEntity(HttpProducer.java:523)
at org.apache.camel.component.http4.HttpProducer.createMethod(HttpProducer.java:422)
at org.apache.camel.component.http4.HttpProducer.process(HttpProducer.java:110)
at org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:468)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
at org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:681)
at org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:609)
at org.apache.camel.processor.MulticastProcessor.process(MulticastProcessor.java:238)
at org.apache.camel.processor.RecipientList.sendToRecipientList(RecipientList.java:170)
at org.apache.camel.processor.RecipientList.process(RecipientList.java:131)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:121)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:83)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:468)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:121)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:83)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:109)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:91)
at amqp.spring.camel.component.SpringAMQPConsumer$RabbitMQMessageListener.onMessage(SpringAMQPConsumer.java:201)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:757)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:680)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access[=11=]1(SimpleMessageListenerContainer.java:93)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:183)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:302)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
at org.springframework.retry.interceptor.StatefulRetryOperationsInterceptor$MethodInvocationRetryCallback.doWithRetry(StatefulRetryOperationsInterceptor.java:162)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:276)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:201)
at org.springframework.retry.interceptor.StatefulRetryOperationsInterceptor.invoke(StatefulRetryOperationsInterceptor.java:137)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:208)
at com.sun.proxy.$Proxy50.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1358)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:661)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1102)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1086)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access00(SimpleMessageListenerContainer.java:93)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1203)
at amqp.spring.camel.component.SpringAMQPConsumer$SpringAMQPExecutorTask.run(SpringAMQPConsumer.java:316)
at amqp.spring.camel.component.SpringAMQPConsumer$SpringAMQPExecutorTask.run(SpringAMQPConsumer.java:316)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.camel.NoTypeConversionAvailableException: No type converter available to convert from type: uk.gov.insolvency.microservices.pojos.Address to the required type: java.io.InputStream with value Address [houseNumber=null, postcode=DY10 4TW, line1=null, line2=null, line3=null, line4=null, locality=null, town=null, county=null]
at org.apache.camel.impl.converter.BaseTypeConverterRegistry.mandatoryConvertTo(BaseTypeConverterRegistry.java:198)
at org.apache.camel.impl.MessageSupport.getMandatoryBody(MessageSupport.java:105)
... 49 common frames omitted
我尝试创建的路线是：
from("spring-amqp:myExchange:getAddress:camel.getAddress?type=topic&durable=true&autodelete=false")
.log("${body}")
.setHeader(Exchange.HTTP_URL,
constant("https://api.getAddress.io"))
.setExchangePattern(ExchangePattern.InOut)
.setHeader("postcode",
simple("${body.postcode}"))
.setHeader("houseNumber",
simple("${body.houseNumber}"))
.log("${headers}")
.recipientList(simple("https4:{{addressmicroservice.address.url}}/${header.postcode}/${header.houseNumber}?{{addressmicroservice.address.key}}={{addressmicroservice.address.password}}&throwExceptionOnFailure=false&bridgeEndpoint=true"))
.process(processor)
.log("${body}");
使用 Camel 2.17.2 和 Java 8.
从你的错误日志中，你需要在端点之间交换消息时转换消息，你可以参考这个Type Converter，或者你可以marshal/unmarshal将消息xml/json替换为xml/json .
使用 camel，您可以像
这样调用 marshal() 来轻松添加 marshal json 功能
from("activemq:My.Queue").
marshal().json().
to("mqseries:Another.Queue");
尝试编组：
from("spring-amqp:myExchange:getAddress:camel.getAddress?type=topic&durable=true&autodelete=false")
.log("${body}")
.setHeader(Exchange.HTTP_URL,
constant("https://api.getAddress.io"))
.setExchangePattern(ExchangePattern.InOut)
.setHeader("postcode",
simple("${body.postcode}"))
.setHeader("houseNumber",
simple("${body.houseNumber}"))
.log("${headers}")
.marshal().json(Gson).convertBodyTo(String.class)
.recipientList(simple("https4:{{addressmicroservice.address.url}}/${header.postcode}/${header.houseNumber}?{{addressmicroservice.address.key}}={{addressmicroservice.address.password}}&throwExceptionOnFailure=false&bridgeEndpoint=true"))
.process(processor)
.log("${body}");
请记住，为此您需要 camel-gson。对于 gradle 使用：
compile('org.apache.camel:camel-gson:2.17.1')
并根据您的骆驼版本进行调整。
我是 Camel 的新手，到目前为止一直玩得很开心。但是，我现在在尝试从 AMQP 读取消息并使用消息中的参数调用 REST 端点时遇到问题。
我的想法是 AMQP 消息包含一个 POJO，我正在阅读它以获得邮政编码和门牌号值。这些值然后用于填充 URL 中的动态字段。 我正在使用 HTTP4 执行对 REST 端点的调用，每次我点击该行时，我都会收到：
Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor Elapsed (ms)
[route1 ] [route1 ] [spring-amqp://myExchange:getAddress:camel.getAddress?autodelete=false&durable=] [ 46]
[route1 ] [log1 ] [log ] [ 13]
[route1 ] [setHeader1 ] [setHeader[CamelHttpUrl] ] [ 0]
[route1 ] [setExchangePattern] [setExchangePattern[InOut] ] [ 0]
[route1 ] [setHeader2 ] [setHeader[postcode] ] [ 18]
[route1 ] [setHeader3 ] [setHeader[houseNumber] ] [ 1]
[route1 ] [log2 ] [log ] [ 0]
[route1 ] [recipientList1 ] [recipientList[simple{Simple: https4:{{addressmicroservice.address.url}}/${head] [ 13]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.InvalidPayloadException: No body available of type: java.io.InputStream but has value: Address [houseNumber=null, postcode=DY10 4TW, line1=null, line2=null, line3=null, line4=null, locality=null, town=null, county=null] of type: uk.gov.insolvency.microservices.pojos.Address on: Message[ID-Andrews-MacBook-Pro-local-56293-1472721966485-0-3]. Caused by: No type converter available to convert from type: uk.gov.insolvency.microservices.pojos.Address to the required type: java.io.InputStream with value Address [houseNumber=null, postcode=DY10 4TW, line1=null, line2=null, line3=null, line4=null, locality=null, town=null, county=null]. Exchange[ID-Andrews-MacBook-Pro-local-56293-1472721966485-0-4]. Caused by: [org.apache.camel.NoTypeConversionAvailableException - No type converter available to convert from type: uk.gov.insolvency.microservices.pojos.Address to the required type: java.io.InputStream with value Address [houseNumber=null, postcode=DY10 4TW, line1=null, line2=null, line3=null, line4=null, locality=null, town=null, county=null]]
at org.apache.camel.impl.MessageSupport.getMandatoryBody(MessageSupport.java:107)
at org.apache.camel.component.http4.HttpProducer.createRequestEntity(HttpProducer.java:523)
at org.apache.camel.component.http4.HttpProducer.createMethod(HttpProducer.java:422)
at org.apache.camel.component.http4.HttpProducer.process(HttpProducer.java:110)
at org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:468)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
at org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:681)
at org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:609)
at org.apache.camel.processor.MulticastProcessor.process(MulticastProcessor.java:238)
at org.apache.camel.processor.RecipientList.sendToRecipientList(RecipientList.java:170)
at org.apache.camel.processor.RecipientList.process(RecipientList.java:131)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:121)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:83)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:468)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:121)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:83)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:109)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:91)
at amqp.spring.camel.component.SpringAMQPConsumer$RabbitMQMessageListener.onMessage(SpringAMQPConsumer.java:201)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:757)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:680)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access[=11=]1(SimpleMessageListenerContainer.java:93)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:183)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:302)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
at org.springframework.retry.interceptor.StatefulRetryOperationsInterceptor$MethodInvocationRetryCallback.doWithRetry(StatefulRetryOperationsInterceptor.java:162)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:276)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:201)
at org.springframework.retry.interceptor.StatefulRetryOperationsInterceptor.invoke(StatefulRetryOperationsInterceptor.java:137)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:208)
at com.sun.proxy.$Proxy50.invokeListener(Unknown Source)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1358)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:661)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1102)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1086)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access00(SimpleMessageListenerContainer.java:93)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1203)
at amqp.spring.camel.component.SpringAMQPConsumer$SpringAMQPExecutorTask.run(SpringAMQPConsumer.java:316)
at amqp.spring.camel.component.SpringAMQPConsumer$SpringAMQPExecutorTask.run(SpringAMQPConsumer.java:316)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.camel.NoTypeConversionAvailableException: No type converter available to convert from type: uk.gov.insolvency.microservices.pojos.Address to the required type: java.io.InputStream with value Address [houseNumber=null, postcode=DY10 4TW, line1=null, line2=null, line3=null, line4=null, locality=null, town=null, county=null]
at org.apache.camel.impl.converter.BaseTypeConverterRegistry.mandatoryConvertTo(BaseTypeConverterRegistry.java:198)
at org.apache.camel.impl.MessageSupport.getMandatoryBody(MessageSupport.java:105)
... 49 common frames omitted
我尝试创建的路线是：
from("spring-amqp:myExchange:getAddress:camel.getAddress?type=topic&durable=true&autodelete=false")
.log("${body}")
.setHeader(Exchange.HTTP_URL,
constant("https://api.getAddress.io"))
.setExchangePattern(ExchangePattern.InOut)
.setHeader("postcode",
simple("${body.postcode}"))
.setHeader("houseNumber",
simple("${body.houseNumber}"))
.log("${headers}")
.recipientList(simple("https4:{{addressmicroservice.address.url}}/${header.postcode}/${header.houseNumber}?{{addressmicroservice.address.key}}={{addressmicroservice.address.password}}&throwExceptionOnFailure=false&bridgeEndpoint=true"))
.process(processor)
.log("${body}");
使用 Camel 2.17.2 和 Java 8.
从你的错误日志中，你需要在端点之间交换消息时转换消息，你可以参考这个Type Converter，或者你可以marshal/unmarshal将消息xml/json替换为xml/json .
使用 camel，您可以像
这样调用 marshal() 来轻松添加 marshal json 功能from("activemq:My.Queue").
marshal().json().
to("mqseries:Another.Queue");
尝试编组：
from("spring-amqp:myExchange:getAddress:camel.getAddress?type=topic&durable=true&autodelete=false")
.log("${body}")
.setHeader(Exchange.HTTP_URL,
constant("https://api.getAddress.io"))
.setExchangePattern(ExchangePattern.InOut)
.setHeader("postcode",
simple("${body.postcode}"))
.setHeader("houseNumber",
simple("${body.houseNumber}"))
.log("${headers}")
.marshal().json(Gson).convertBodyTo(String.class)
.recipientList(simple("https4:{{addressmicroservice.address.url}}/${header.postcode}/${header.houseNumber}?{{addressmicroservice.address.key}}={{addressmicroservice.address.password}}&throwExceptionOnFailure=false&bridgeEndpoint=true"))
.process(processor)
.log("${body}");
请记住，为此您需要 camel-gson。对于 gradle 使用：
compile('org.apache.camel:camel-gson:2.17.1')
并根据您的骆驼版本进行调整。