java - 获取具有属性的 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