基于AMQP的Spring Boot JmsListener在TextMessage上失败

Question

我有一个Spring Boot应用程序，它在从ActiveMQ代理检索类型为TextMessage的JMS消息时遇到问题。

如果消费者尝试从代理检索消息，它无法自动将消息转换为TextMessage，而是将其视为ByteMessage。有一个JmsListener应该将队列中的消息作为TextMessage读取：

...
@JmsListener(destination = "foo")
public void jmsConsumer(TextMessage message) {
...

JmsListener产生如下警告，并删除消息：

org.springframework.jms.listener.adapter.ListenerExecutionFailedException: Listener method could not be invoked with incoming message
Endpoint handler details:
Method [public void net.aschemann.demo.springboot.jmsconsumer.JmsConsumer.jmsConsumer(javax.jms.TextMessage)]
Bean [net.aschemann.demo.springboot.jmsconsumer.JmsConsumer@4715f07]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [javax.jms.TextMessage] for org.springframework.jms.listener.adapter.AbstractAdaptableMessageListener$MessagingMessageConverterAdapter$LazyResolutionMessage@7c49d298, failedMessage=org.springframework.jms.listener.adapter.AbstractAdaptableMessageListener$MessagingMessageConverterAdapter$LazyResolutionMessage@7c49d298
    at org.springframework.jms.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:118) ~[spring-jms-5.1.4.RELEASE.jar:5.1.4.RELEASE]

我提取了一个小样本应用程序来调试该问题：https://github.com/ascheman/springboot-camel-jms

现实生活中的生产者是一个利用Apache Camel的商业应用程序。因此，我几乎无法更改/定制制片人。我试图建立一个样本生产者，显示出同样的行为。

我可以以某种方式调整消费者以将消息视为TextMessage吗？

此外：有没有办法直接在Spring中通过编程从消息中检索额外的AMQP属性？当然，我仍然可以将消息读取为ByteMessage，并尝试解析属性。但我正在寻找一种更干净的方式，它由任何Spring API支持。到目前为止，Spring标题注释没有起到任何作用。

Answer 1

如果从字节[]到字符串的转换有问题，请使用：

.convertBodyTo(String.class)

路线示例：

from(QUEUE_URL)
                .routeId("consumer")
                .convertBodyTo(String.class)
                .log("${body}")
                .to("mock:mockRoute");

Answer 2

我有同样的错误，这是因为LazyResolutionMessage从MessagingMessageConverter调用，这是MessageConverter的默认实现，它转换您的消息（实际上它没有，因为它是默认的）：

return ((org.springframework.messaging.Message) payload).getPayload();

我已经实现了你想要的，最终我的消费者的工作方式是：

@JmsListener(destination = "${someName}")
public void consumeSomeMessages(MyCustomEvent e) {
  ....
}

我要做的是：

@Bean(name = "jmsListenerContainerFactory")
public DefaultJmsListenerContainerFactory whateverNameYouWant(final ConnectionFactory genericCF) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setErrorHandler(t -> log.error("bad consumer, bad", t));
    factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
    factory.setConnectionFactory(genericCF);
    factory.setMessageConverter(
            new MessageConverter() {
                @Override
                public Message toMessage(Object object, Session session) {
                    throw new UnsupportedOperationException("since it's only for consuming!");
                }

                @Override
                public MyCustomEvent fromMessage(Message m) {
                    try {
                      // whatever transformation you want here...
                      // here you could print the message, try casting,
                      // building new objects with message's attributes, so on...
                      // example:
                      return (new ObjectMapper()).readValue(((TextMessage) m).getText(), MyCustomEvent.class);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }
            }
    );
    return factory;
}

几个关键点：

>

请注意，在尝试转换/转换消息类型时，您还可以实现一个ErrorHandler来处理异常！

因为我想从SQS队列中消费，所以ConnectionFactory是一个Spring管理的bean，有Amazon的SQSConnectionFactory。请正确提供您的等价物。我的是：

@Bean("connectionFactory")
public SQSConnectionFactory someOtherNome() {
    return new SQSConnectionFactory(
            new ProviderConfiguration(),
            AmazonSQSClientBuilder.standard()
                    .withRegion(Regions.US_EAST_1)
                    .withCredentials(
                            new AWSStaticCredentialsProvider(
                                    new BasicAWSCredentials(
                                            "keyAccess",
                                            "keySecret"
                                    )
                            )
                    )
                    .build()
    );
}

Answer 3

在我跟随@AndyWilkinson的评论添加了transport之后，我曾与问题所有者面临过同样的问题。activemq中transportConnector上的transformer选项。如下图所示，问题得到了解决。

<transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600&amp;transport.transformer=jms"/>