获取具有属性的 RabbitMQ 主体

Question

我正在像这样从 RabbitMQ 读取：connection = factory.newConnection();ch = connection.createChannel() ;String queueName = managerProps.getProperty("rmq.queue.name");ch.queueDeclare(queueName ,true,false,false, null) ;while (true) {    GetResponse chResponse = ch.basicGet(queueName, false);    logger.info("----" + new String(chResponse.getBody(), "UTF-8") + " ---\n");}这是我在日志中看到的：[Thread-5] INFO com.mycompany.RmqReader - ----?? ♣wx .com.rabbitmq.jms.client.message.RMQTextMessage $b1213c86-10f4-4113-bd2f-45aaabce083f   ♠ ←rmq.jms.meamqpQueueNameq ~ ☺L ♫amqpRoutingKeyq ~ ☺L ☼destinationNameq ~ ☺xp ☺ t ↕jms.durable.queuest !MY.Queue.Name ~ ♦q ~ ♦z  ☻O ↔rmq.jms.message.delivery.mode♦   ☻ ↓rmq.jms.message.timestamp♣  ☺j?∟ ↑rmq.jms.message.priority♦   ♦ →rmq.jms.message.expiration♣         ↕rmq.jms.message.i 'ID:b1213c86-10f4-4113-bd2f-45aaabce083f   ♂ ◄objectTransaction☺☺ ►templateEndpoin -jtemplate://JSONDeliveryTemplateParallel.java ►deliveryLocatio +jms:queue:My.Queue.Name ►destinationIndex♦   ☺ ♫subsCutOffTime♦     ♀breadcrumbI ♀1149808347.0 ◄globalDeliveryUID♦ ?]? ►subscriptionNam §option_session_pubsub ◄originalMessageI ♀1149808347.0 ↨subscriptionDeliveryUID♦ ??8 ¶transactionTimestamp♣  ☺j??(z  ♥R   ♥M[{"OptSession": {.... the actual body is here....}}] ---为什么我在这里看到标题？我如何实际提取身体？

Answer 1

看起来您已经使用 RMQConnectionFactory 发布了符合 jms 的数据，现在您正在使用非 jms ConnectionFactory

请查看使用 RMQConnectionFactory 的示例消费者https://github.com/kunhaj/samples/blob/master/rabbitmq/src/main/java/RabbitMqConsumer.java

import com.rabbitmq.jms.admin.RMQConnectionFactory;

import javax.jms.*;

/**

 *  docker run -d --hostname my-rabbit --name 

 *   some-rabbit  -p 5672:5672 -p 15672:15672 rabbitmq:3-management

 */

public class RabbitMqConsumer {

    public static void main(String[] args) throws Exception {

        RMQConnectionFactory connectionFactory = new RMQConnectionFactory();

        connectionFactory.setUsername("guest");

        connectionFactory.setPassword("guest");

        connectionFactory.setVirtualHost("/");

        connectionFactory.setHost("localhost");

        connectionFactory.setPort(5672);

        connectionFactory.setDeclareReplyToDestination(false);

        Connection connection = connectionFactory.createConnection();

        connection.start();

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        Queue queue = session.createQueue("TEST-QUEUE");

        drainQueue(session, queue);

    }

    protected static void drainQueue(Session session, Queue queue) throws Exception {

        MessageConsumer receiver = session.createConsumer(queue);

        Message msg = receiver.receiveNoWait();

        while (msg != null) {

            String msgBody = ((TextMessage) msg).getText();

            System.out.println("recieved" + msgBody);

            msg = receiver.receiveNoWait();

        }

    }

}

另请参阅 JMS 和 AMQP 0-9-1 目标互操作性https://www.rabbitmq.com/jms-client.html#destination-interoperability