无法使用 Qpid JMS 向 ActiveMQ Artemis 发送大消息

Question

我尝试从磁盘读取一个大文件并将其发送到 ActiveMQ Artemis 上的队列。每次我尝试时都会遇到相同的异常：

Caused by: jakarta.jms.MessageFormatException: Only objectified primitive objects and String types are allowed but was: java.io.BufferedInputStream@3a8489ac type: class java.io.BufferedInputStream
    at org.apache.qpid.jms.message.JmsMessagePropertySupport.checkValidObject(JmsMessagePropertySupport.java:121)
    at org.apache.qpid.jms.message.JmsMessagePropertyIntercepter.setProperty(JmsMessagePropertyIntercepter.java:725)
    at org.apache.qpid.jms.message.JmsMessage.setObjectProperty(JmsMessage.java:348)
    at org.apache.camel.component.jms.JmsBinding.createJmsMessageForType(JmsBinding.java:661)
    at org.apache.camel.component.jms.JmsBinding.createJmsMessage(JmsBinding.java:568)
    at org.apache.camel.component.jms.JmsBinding.createJmsMessage(JmsBinding.java:525)
    at org.apache.camel.component.jms.JmsBinding.makeJmsMessage(JmsBinding.java:348)
    at org.apache.camel.component.jms.JmsProducer$2.createMessage(JmsProducer.java:325)
    at org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemplate.doSendToDestination(JmsConfiguration.java:616)
    at org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemplate.lambda$send$0(JmsConfiguration.java:574)
    at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:507)
    ... 39 more

在我看来，这永远不会起作用，因为在调试代码时，我发现一个地方设置了库的属性

message.setObjectProperty("JMS_AMQ_InputStream", is);

，并且该属性永远不会有效，因为它是一个

InputStream

。知道如何解决这个问题吗？

    if (endpoint.isArtemisStreamingEnabled()) {
        LOG.trace("Optimised for Artemis: Streaming payload in BytesMessage");
        InputStream is = context.getTypeConverter().mandatoryConvertTo(InputStream.class, exchange, body);
        message.setObjectProperty("JMS_AMQ_InputStream", is);
        LOG.trace("Optimised for Artemis: Finished streaming payload in BytesMessage");
    } else {

来源

public static void setProperty(JmsMessage message, String name, Object value) throws JMSException {
    PropertyIntercepter jmsPropertyExpression = PROPERTY_INTERCEPTERS.get(name);

    if (jmsPropertyExpression == null || !jmsPropertyExpression.isAlwaysWritable()) {
        message.checkReadOnlyProperties();
    }
    checkPropertyNameIsValid(name, message.isValidatePropertyNames());
    checkValidObject(value);

    if (jmsPropertyExpression != null) {
        jmsPropertyExpression.setProperty(message, value);
    } else {
        message.getFacade().setProperty(name, value);
    }
}

来源

public static void checkValidObject(Object value) throws MessageFormatException {
    boolean valid = value instanceof Boolean ||
                    value instanceof Byte ||
                    value instanceof Short ||
                    value instanceof Integer ||
                    value instanceof Long ||
                    value instanceof Float ||
                    value instanceof Double ||
                    value instanceof Character ||
                    value instanceof String ||
                    value == null;

    if (!valid) {
        throw new MessageFormatException("Only objectified primitive objects and String types are allowed but was: " + value + " type: " + value.getClass());
    }
}

来源

Answer 1

JMS_AMQ_InputStream

属性是独占用于核心JMS客户端实现。根据堆栈跟踪，您尝试将此属性与 Qpid JMS 客户端t 实现一起使用，但这是行不通的。

您需要更改类路径以引用 Core JMS 客户端库，或者停止使用

JMS_AMQ_InputStream

属性。