骆驼处理来自 RabbitMQ 的 JSON 消息

Question

我想将 JSON 格式的消息发布到 RabbitMQ 并成功使用该消息。我正在尝试使用 Camel 来整合生产者和消费者。但是，我很难理解如何创建一条路线来实现这一点。我使用 JSON Schema 来定义生产者和消费者之间的接口。我的应用程序创建 JSON，将其转换为 byte[] 并使用 Camel ProducerTemplate 将消息发送到 RabbitMQ。在消费者端，需要将 byte[] 消息转换为 String，然后转换为 JSON，然后编组为 Object，以便我可以处理它。但是以下代码行不起作用

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

就好像 bean 传递了原始 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、Annotations 等，希望服务激活尽可能简单。尽量使用骆驼

这是制作人

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，以便可以使用 setter 方法，并添加了一个 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 处理程序中的故障正确备份堆栈