我正在像這樣從 RabbitMQ 讀?。篶onnection = 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....}}] ---為什么我在這里看到標題？我如何實際提取身體？
1 回答
當年話下
TA貢獻1890條經驗 獲得超9個贊
看起來您已經使用 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
添加回答
舉報
0/150
提交
取消