Как извлечь тело сообщения из потребительского тега с помощью DefaultConsumer? - PullRequest
Новая
       15

Как извлечь тело сообщения из потребительского тега с помощью DefaultConsumer?

0 голосов
/ 07 мая 2018

Я пытаюсь создать простое приложение с RabbitMQ Publisher and Consumer, производитель написан на Java, а потребитель - на Scala.

Это используемый мной издатель, который успешно вставляет данные в очередь «queue1» на сервере rabbitMQ: 

String queue="queue1";


String exchange="queue1";
String routing_key="queue1";
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(queue, true, false, false, null);
channel.exchangeDeclare(exchange, "direct");
channel.queueBind(queue,exchange,routing_key);
channel.basicQos(1);

String msg = "Hello";

channel.basicPublish(exchange,routing_key_one, new AMQP.BasicProperties.Builder().contentType("text/plain").deliveryMode(1).priority(0).
build(),msg.getBytes());
channel.close();
connection.close();

Проблема сейчас в том, что у меня есть Потребитель по отдельному сценарию, но я не могу заставить его вернуть строку "msg", которую я отправил от издателя; Я могу получить только потребительский тег в качестве возвращаемого значения. Это мой потребитель: 

import com.rabbitmq.client.{ConnectionFactory,Connection,Channel,Consumer,DefaultConsumer,Envelope,BasicProperties}

val rabbit_host="localhost"
val queue="queue1"
val exchange="queue1"
val routing_key="queue1"

val factory: ConnectionFactory = new ConnectionFactory()
factory.setHost (rabbit_host)

val connection: Connection = factory.newConnection()
val channel: Channel = connection.createChannel()

channel.queueBind(queue_name, exchange, routing_key, null)
channel.basicQos(1)

val consumer:Consumer=new DefaultConsumer(channel){
  def handleDelivery(consumerTag: String, envelope: Envelope, properties: BasicProperties, body: Array[Byte]):String ={
    new String(body,"UTF-8")
  }
}


val msg: String = channel.basicConsume(queue_name,true, consumer)


channel.close()
connection.close()

Если я затем напечатаю «msg», это значение, которое я получаю: «amq.ctag-miRiLnyTcsq9MwzHyVshZw»

Итак, мой вопрос: есть ли способ получить "привет" (исходное значение) в качестве возврата для функции basicConsume?

Спасибо за ваше время.

1 Ответ

0 голосов
/ 08 мая 2018

Согласно документации 

val msg: String = channel.basicConsume(queue_name,true, consumer)

Возвращает

тег потребителя, связанный с новым потребителем

Из-за асинхронного характера rabbitMq ждать сообщения нелегко. Вы можете попробовать использовать Promise и заблокировать результат 

val p = Promise[String]()
val f = p.future

val consumer:Consumer=new DefaultConsumer(channel){
  def handleDelivery(consumerTag: String, envelope: Envelope, properties: BasicProperties, body: Array[Byte]):String ={
    val parsedBody = new String(body,"UTF-8")
    p.success(parsedBody)
  }
}
val msg: String = Await.result(f, 5 seconds)
...