无效的流标头-在rabbitmq控制台上手动发布消息时

Question

我有一个带有spring-jms、spring-boot和rabbitmq-jms的应用程序。

当我尝试发送消息并通过应用程序成功接收时。但当我在rabbitmqconsole上手动发布消息时，我在应用程序侦听时出现了以下错误：

com.rabbitmq.jms.util.RMQJMSException: invalid stream header: 61736466
at com.rabbitmq.jms.client.RMQMessage.fromMessage(RMQMessage.java:1140) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
at com.rabbitmq.jms.client.RMQMessage.convertJmsMessage(RMQMessage.java:913) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
at com.rabbitmq.jms.client.RMQMessage.convertMessage(RMQMessage.java:907) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
at com.rabbitmq.jms.client.RMQMessageConsumer.receive(RMQMessageConsumer.java:356) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
at com.rabbitmq.jms.client.RMQMessageConsumer.receive(RMQMessageConsumer.java:269) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
at org.springframework.jms.support.destination.JmsDestinationAccessor.receiveFromConsumer(JmsDestinationAccessor.java:132) ~[spring-jms-5.3.13.jar:5.3.13]
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveMessage(AbstractPollingMessageListenerContainer.java:418) ~[spring-jms-5.3.13.jar:5.3.13]
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:303) ~[spring-jms-5.3.13.jar:5.3.13]
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:257) ~[spring-jms-5.3.13.jar:5.3.13]
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1237) ~[spring-jms-5.3.13.jar:5.3.13]
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1227) ~[spring-jms-5.3.13.jar:5.3.13]
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1120) ~[spring-jms-5.3.13.jar:5.3.13]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
Caused by: java.io.StreamCorruptedException: invalid stream header: 61736466
at java.base/java.io.ObjectInputStream.readStreamHeader(ObjectInputStream.java:940) ~[na:na]
at java.base/java.io.ObjectInputStream.<init>(ObjectInputStream.java:379) ~[na:na]
at com.rabbitmq.jms.util.WhiteListObjectInputStream.<init>(WhiteListObjectInputStream.java:90) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
at com.rabbitmq.jms.client.RMQMessage.fromMessage(RMQMessage.java:1114) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
... 12 common frames omitted

这是我的JMS配置

@EnableJms
@Configuration
public class ConnectionRabbitConfig {
@Autowired RabbitProperties rabbitProperties;
@Bean
public ConnectionFactory jmsConnectionFactory() {
RMQConnectionFactory connectionFactory = new RMQConnectionFactory();
connectionFactory.setUsername(rabbitProperties.getUser());
connectionFactory.setPassword(rabbitProperties.getPass());
connectionFactory.setVirtualHost(rabbitProperties.getVirtualhost());
connectionFactory.setHost(rabbitProperties.getHost());
connectionFactory.setPort(rabbitProperties.getPort());
return connectionFactory;
}

@Bean
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(@Autowired ConnectionFactory connectionFactory) {
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setAutoStartup(rabbitProperties.getListenerEnabled());
return factory;
}
@Bean
public JmsTemplate defaultJmsTemplate(@Autowired ConnectionFactory connectionFactory) {
return new JmsTemplate(connectionFactory);
}
}

发布消息

@Autowired
private JmsTemplate jmsTemplate;
public void publish(String queueName, String message, Map<String, Object> headers) {
jmsTemplate.convertAndSend(queueName, message );
}

消费信息

@JmsListener(destination = "queue-1")
public void consume(@Payload Message message,  @Headers MessageHeaders headers) {
// runs successfully if I listen message added by same application
// but if I try to add manualy on rabbit console I got error.
}

libs

implementation 'org.springframework:spring-jms:5.3.13'
implementation 'com.rabbitmq.jms:rabbitmq-jms:2.3.0'

为什么我需要在rabbit控制台上添加属性或标头才能成功侦听应用程序？或者我需要更改配置代码吗？

Answer 1

只是在这里添加我的解决方案。

默认情况下，spring将我的侦听器添加到具有non-ampq的队列上的交换机jms.durable.queues中

因为这是在应用程序发布时工作，但在控制台上添加TextMessage时抛出错误。控制台上发布的所有消息都是一条非ampq消息，在springjms客户端lib上因无效标头而发生冲突。

解决方案：

我添加了一个目标解析程序，它连接到amqp上的队列，而不创建默认的交换jms.durable.queues此处提供解决方案的更多详细信息：https://stackoverflow.com/a/70269301/7505687