Spring Cloud Stream读取Oracle AQ自定义ADT负载报错JMS-137求助

问题根源 #

报错JMS-137: Payload factory must be specified for destinations with ADT payloads是因为Spring JMS容器创建Consumer时，未为Oracle AQ的ADT队列绑定对应的ORADataFactory（即你实现的MessageContainer工厂类）。纯Java方式可行是因为手动处理了工厂绑定，但Spring自动配置未覆盖该场景。

关键修改点 #

1. 修正ConnectionFactory配置：临时连接设置typeMap后关闭的操作无法作用于池化连接，需将typeMap配置到OracleDataSource本身，确保所有新连接继承映射关系。
2. 自定义JmsListenerContainerFactory：通过扩展Spring容器工厂，在创建Consumer时为ADT队列指定PayloadFactory。

修改后的代码 #

OracleAQConfig.java #

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Map;
import java.util.Properties;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.connection.CachingConnectionFactory;

import jakarta.jms.ConnectionFactory;
import jakarta.jms.JMSException;
import oracle.jakarta.jms.AQjmsDestination;
import oracle.jakarta.jms.AQjmsFactory;
import oracle.jdbc.pool.OracleDataSource;

@Configuration
public class OracleAQConfig {

    @Bean
    public ConnectionFactory connectionFactory() throws JMSException, SQLException {
        Properties props = new Properties();
        props.setProperty("oracle.jms.mapMessage", "true");

        OracleDataSource dataSource = new OracleDataSource();
        dataSource.setURL("jdbc:oracle:thin:@//dbaddress.com:1521/MYSID");
        dataSource.setUser("jmsuser");
        dataSource.setPassword("jmspassword");
        dataSource.setConnectionProperties(props);

        // 为数据源设置默认TypeMap，确保所有池化连接继承该映射
        try (Connection conn = dataSource.getConnection()) {
            Map<String, Class<?>> typeMap = conn.getTypeMap();
            typeMap.put(MessageContainer.SQL_NAME, MessageContainer.class);
            dataSource.setTypeMap(typeMap);
        } catch (Exception e) {
            throw new RuntimeException("Error setting up custom ADT type map", e);
        }

        // 缓存连接提升性能，可选但推荐
        CachingConnectionFactory cachingConnFactory = new CachingConnectionFactory();
        cachingConnFactory.setTargetConnectionFactory(AQjmsFactory.getConnectionFactory(dataSource));
        cachingConnFactory.setSessionCacheSize(10);
        return cachingConnFactory;
    }

    @Bean(name = "jmsListenerContainerFactory")
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrency("1-5");
        
        // 自定义Consumer创建逻辑，绑定ADT对应的PayloadFactory
        factory.setConsumerCustomizer((session, consumer, destination) -> {
            if (destination instanceof AQjmsDestination aqDestination) {
                try {
                    aqDestination.setPayloadFactory(MessageContainer.getORADataFactory());
                } catch (JMSException e) {
                    throw new RuntimeException("Failed to set payload factory for ADT destination", e);
                }
            }
        });
        
        return factory;
    }
}

生效原因 #

• 全局TypeMap配置：将ADT与Java类的映射设置到OracleDataSource，确保每个池化连接都能识别ADT类型。
• ConsumerCustomizer扩展：通过Spring的扩展点，在创建Consumer前为Oracle AQ目标绑定PayloadFactory，满足Oracle对ADT消息的处理要求。

额外验证项 #

• 确认MessageContainer的SQL_NAME完全匹配数据库ADT名称（包含用户名前缀jmsuser.MESSAGE_CONTAINER）。
• 检查Oracle JDBC与AQ API版本兼容性（当前23.3版本组合兼容）。

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