骆驼处理来自RabbitMQ的JSON消息

Question

我希望以JSON格式向RabbitMQ发布一条消息，并成功地使用该消息。我试图用骆驼来整合生产者和消费者。然而，我很难理解如何创建一条实现这一目标的途径。我使用JSON来定义生产者和消费者之间的接口。我的应用程序创建JSON，将其转换为byte[]，并使用Camel ProducerTemplate将消息发送到RabbitMQ。在使用者端，需要将byte[]消息转换为字符串，然后转换为JSON，然后封送到对象，以便我可以处理它。但是，下面的代码行不起作用

from(startEndpoint).transform(body().convertToString()).marshal().json(JsonLibrary.Jackson, classOf[Payload]).bean(classOf[JsonBeanExample]), 

这就好像传递的是原始的byte[]内容，而不是JSON json(JsonLibrary.Jackson, classOf[Payload])创建的对象。我看到的所有使用json(..)的骆驼例子调用似乎后面跟着一个to(.)路线的终点是哪一条？这是错误消息

Caused by: org.apache.camel.InvalidPayloadException: No body available of type: uk.co.techneurons.messaging.Payload but has value: [B@48898819 of type: byte[] on: Message: "{\"id\":1}". Caused by: No type converter available to convert from type: byte[] to the required type:     uk.co.techneurons.messaging.Payload with value [B@48898819. Exchange[ID-Tonys-    iMac-local-54996-1446407983661-0-2][Message: "{\"id\":1}"]. Caused by:     [org.apache.camel.NoTypeConversionAvailableException - No type converter available to convert from type: byte[] to the required type: uk.co.techneurons.messaging.Payload with value [B@48898819]`    

我真的不想使用Spring，注解等，想要尽可能简单的服务激活。尽可能多地使用骆驼

这位是制片人

package uk.co.techneurons.messaging
import org.apache.camel.builder.RouteBuilder
import org.apache.camel.impl.DefaultCamelContext

object RabbitMQProducer extends App {
   val camelContext = new DefaultCamelContext
   val rabbitMQEndpoint: String = "rabbitmq:localhost:5672/advert?autoAck=false&threadPoolSize=1&username=guest&password=guest&exchangeType=topic&autoDelete=false&declare=false"
   val rabbitMQRouteBuilder = new RouteBuilder() {
     override def configure(): Unit = {
       from("direct:start").to(rabbitMQEndpoint)
     }
   }
   camelContext.addRoutes(rabbitMQRouteBuilder)
   camelContext.start
   val producerTemplate = camelContext.createProducerTemplate
   producerTemplate.setDefaultEndpointUri("direct:start")
   producerTemplate.sendBodyAndHeader("{\"id\":1}","rabbitmq.ROUTING_KEY","advert.edited")
  camelContext.stop
}

这是消费者。

package uk.co.techneurons.messaging
import org.apache.camel.builder.RouteBuilder
import org.apache.camel.impl.DefaultCamelContext
import org.apache.camel.model.dataformat.JsonLibrary

object RabbitMQConsumer extends App {
  val camelContext = new DefaultCamelContext
  val startEndpoint = "rabbitmq:localhost:5672/advert?queue=es_index&exchangeType=topic&autoDelete=false&declare=false&autoAck=false"
  val consumer = camelContext.createConsumerTemplate
  val routeBuilder = new RouteBuilder() {
    override def configure(): Unit = {
        from(startEndpoint).transform(body().convertToString()).marshal().json(JsonLibrary.Jackson, classOf[Payload]).bean(classOf[JsonBeanExample])
    }
 }
 camelContext.addRoutes(routeBuilder)
 camelContext.start
 Thread.sleep(1000)
 camelContext.stop
}

case class Payload(id: Long)

class JsonBeanExample {
   def process(payload: Payload): Unit = {
     println(s"JSON ${payload}")
   }
}

为了完整起见，这是sbt文件，以便于复制。

name := """camel-scala"""

version := "1.0"

scalaVersion := "2.11.7"

libraryDependencies ++= {
 val scalaTestVersion = "2.2.4"
 val camelVersion: String = "2.16.0"
 val rabbitVersion: String = "3.5.6"
 val slf4jVersion: String = "1.7.12"
 val logbackVersion: String = "1.1.3"
 Seq(
   "org.scala-lang.modules" %% "scala-xml" % "1.0.3",
   "org.apache.camel" % "camel-core" % camelVersion,
   "org.apache.camel" % "camel-jackson" % camelVersion,
   "org.apache.camel" % "camel-scala" % camelVersion,
   "org.apache.camel" % "camel-rabbitmq" % camelVersion,
   "com.rabbitmq" % "amqp-client" % rabbitVersion,
   "org.slf4j" % "slf4j-api" % slf4jVersion,
   "ch.qos.logback" % "logback-classic" % logbackVersion,
   "org.apache.camel" % "camel-test" % camelVersion % "test",
   "org.scalatest" %% "scalatest" % scalaTestVersion % "test")
}

谢谢

Answer 1

我决定创建一个Bean并注册它(说起来容易做起来难！--由于某些未知的原因，JNDIRegistry没有使用DefaultCamelContext --所以我使用了SimpleRegistry)，

  val registry: SimpleRegistry  = new SimpleRegistry()
  registry.put("myBean", new JsonBeanExample())
  val camelContext = new DefaultCamelContext(registry)

然后，我改变了消费的routeBuilder --似乎我已经过度地转换信息了。

  from(startEndpoint).unmarshal.json(JsonLibrary.Jackson, classOf[Payload]).to("bean:myBean?method=process")

我还更改了Bean方法，并添加了一个toString

class Payload {
   @BeanProperty var id: Long = _
   override def toString = s"Payload($id)"
} 
class JsonBeanExample() {
  def process(payload: Payload): Unit = {
     println(s"recieved ${payload}")
  }
}

现在的下一个问题是使死信队列正常工作，并确保Bean处理程序中的故障能够正确地备份堆栈