Camel 处理来自 RabbitMQ 的 JSON 条消息

Question

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

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

就好像这个bean传递的是原始的byte[]内容，而不是JSONjson(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，注解等，希望服务激活越简单越好。尽量使用Camel

这是制作人

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 处理程序中的故障能够正确地备份堆栈

Camel 处理来自 RabbitMQ 的 JSON 条消息

Camel Processing JSON Messages from RabbitMQ

java

json

scala

apache-camel

rabbitmq