Spring Integration中JMS与RabbitMQ消息属性映射方案咨询

咱们先拆解你的问题，一步步来梳理：

一、当前实现的正确性与返回类型说明 #

你的实现方向是对的，用ServiceActivator来做跨中间件的消息属性映射完全可行。关于返回类型：

返回org.springframework.amqp.core.Message是完全正确的，因为<int-amqp:outbound-channel-adapter>可以直接接收这个类型的消息，不需要额外配置转换器——出站适配器会直接将这个Message对象发送到RabbitMQ。

不过你当前的代码有个关键疏漏：只初始化了MessageProperties，但没有处理消息体，也没有创建并返回完整的Message实例。这会导致出站适配器无法拿到有效消息内容，必须补上这部分逻辑。

二、完整的JMS（MQSeries）到RabbitMQ属性映射实现 #

要覆盖所有RabbitMQ标准属性的映射，我们需要处理消息体转换+标准属性映射+自定义JMS属性映射这三部分。下面是修正后的完整代码：

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import javax.jms.*;
import java.nio.charset.StandardCharsets;
import java.util.Enumeration;

public class ServiceActivator {

    public Message convertMessageMQSeriesToRabbit(javax.jms.Message jmsMessage) {
        MessageProperties rabbitProps = new MessageProperties();
        byte[] messageBody = new byte[0];

        try {
            // 1. 处理消息体：根据JMS消息类型转换为RabbitMQ的字节数组
            if (jmsMessage instanceof TextMessage) {
                String textBody = ((TextMessage) jmsMessage).getText();
                messageBody = textBody.getBytes(StandardCharsets.UTF_8);
                rabbitProps.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
            } else if (jmsMessage instanceof BytesMessage) {
                BytesMessage bytesMsg = (BytesMessage) jmsMessage;
                messageBody = new byte[(int) bytesMsg.getBodyLength()];
                bytesMsg.readBytes(messageBody);
                rabbitProps.setContentType(MessageProperties.CONTENT_TYPE_OCTET_STREAM);
            }

            // 2. 标准属性映射
            rabbitProps.setCorrelationId(jmsMessage.getJMSCorrelationID());
            rabbitProps.setPriority(jmsMessage.getJMSPriority());
            rabbitProps.setType(jmsMessage.getJMSType());
            // 处理content_encoding：如果JMS有自定义属性，直接映射
            if (jmsMessage.propertyExists("content_encoding")) {
                rabbitProps.setContentEncoding(jmsMessage.getStringProperty("content_encoding"));
            }

            // 3. 自定义JMS属性映射到RabbitMQ的headers
            Enumeration<?> propertyNames = jmsMessage.getPropertyNames();
            while (propertyNames.hasMoreElements()) {
                String propName = (String) propertyNames.nextElement();
                Object propValue = jmsMessage.getObjectProperty(propName);
                // 把所有自定义JMS属性放到Rabbit的headers中，方便后续处理
                rabbitProps.getHeaders().put(propName, propValue);
            }

        } catch (JMSException e) {
            e.printStackTrace();
            // 抛出异常让Spring Integration的错误通道处理，避免静默失败
            throw new RuntimeException("Failed to convert MQSeries message to RabbitMQ message", e);
        }

        // 创建并返回完整的RabbitMQ Message对象
        return new Message(messageBody, rabbitProps);
    }
}

三、RabbitMQ到MQSeries的反向属性映射实现 #

要实现反向转换，我们需要编写另一个ServiceActivator方法，将RabbitMQ消息转换为JMS消息。注意这里可以让Spring Integration自动注入JMS Session，用来创建JMS消息实例：

import org.springframework.amqp.core.Message;
import javax.jms.*;
import java.nio.charset.StandardCharsets;

public class ServiceActivator {

    // 前面的convertMessageMQSeriesToRabbit方法...

    public javax.jms.Message convertMessageRabbitToMQSeries(Message rabbitMessage, Session jmsSession) throws JMSException {
        MessageProperties rabbitProps = rabbitMessage.getMessageProperties();
        javax.jms.Message jmsMessage;

        // 1. 根据RabbitMQ的contentType创建对应的JMS消息类型
        if (MessageProperties.CONTENT_TYPE_TEXT_PLAIN.equals(rabbitProps.getContentType())) {
            String textBody = new String(rabbitMessage.getBody(), StandardCharsets.UTF_8);
            jmsMessage = jmsSession.createTextMessage(textBody);
        } else {
            BytesMessage bytesMsg = jmsSession.createBytesMessage();
            bytesMsg.writeBytes(rabbitMessage.getBody());
            jmsMessage = bytesMsg;
        }

        // 2. 标准属性反向映射
        jmsMessage.setJMSCorrelationID(rabbitProps.getCorrelationId());
        jmsMessage.setJMSPriority(rabbitProps.getPriority());
        jmsMessage.setJMSType(rabbitProps.getType());
        // 映射content_encoding到JMS自定义属性
        if (rabbitProps.getContentEncoding() != null) {
            jmsMessage.setStringProperty("content_encoding", rabbitProps.getContentEncoding());
        }

        // 3. RabbitMQ的headers映射到JMS自定义属性
        rabbitProps.getHeaders().forEach((key, value) -> {
            try {
                // 注意JMS属性只支持基本类型、String和Serializable对象，需要做类型兼容处理
                if (value instanceof String || value instanceof Number || value instanceof Boolean) {
                    jmsMessage.setObjectProperty(key, value);
                }
            } catch (JMSException e) {
                e.printStackTrace();
            }
        });

        return jmsMessage;
    }
}

对应的反向配置示例：

<int:channel id="rabbitInput" ></int:channel>
<int-amqp:inbound-channel-adapter id="amqpIn" channel="rabbitInput" connection-factory="rabbitConnectionFactory" queue-names="your-rabbit-queue"/>

<int:service-activator id="rabbitToMqSa" input-channel="rabbitInput" ref="serviceBean" method="convertMessageRabbitToMQSeries" output-channel="jmsOutput"/>

<int:channel id="jmsOutput" ></int:channel>
<int-jms:outbound-channel-adapter channel="jmsOutput" connection-factory="connectionFactoryCaching" destination="your-mqseries-queue"/>

额外注意事项 #

• 异常处理：不要只打印栈轨迹，抛出RuntimeException可以触发Spring Integration的错误通道机制，方便你监控和处理转换失败的情况。
• 类型兼容性：JMS自定义属性对类型有限制，映射RabbitMQ headers时要过滤掉不兼容的类型，避免抛出异常。
• 自定义属性策略：如果不需要把所有JMS属性都映射到RabbitMQ，可以根据业务需求只保留关键属性，减少不必要的开销。

内容的提问来源于stack exchange，提问作者Eric NICOLAS