如何在Java提交时排队入Oracle AQ表并通过JMS客户端使用

Question

我正在为企业级产品编写Java组件，并希望利用Oracle 11g数据库的特定功能Active Queues。我要完成的确切场景是-1.在提交时向oracle活动队列/队列表中写入一条消息2.使用JMS使用者从队列中读取该消息

我在http://docs.oracle.com/cd/B28359_01/java.111/b31224/streamsaq.htm上观看了演示和教程。

特别是，我想重点介绍代码的入队部分-

    // Create the actual AQMessage instance:
    AQMessage mesg = AQFactory.createAQMessage(msgprop);
    // and add a payload:
    byte[] rawPayload = new byte[500];
    for (int i = 0; i < rawPayload.length; i++) {
        rawPayload[i] = 'b';
    }

    mesg.setPayload(new RAW(rawPayload));

    AQEnqueueOptions opt = new AQEnqueueOptions();
    opt.setRetrieveMessageId(true);
    opt.setDeliveryMode(AQEnqueueOptions.DeliveryMode.PERSISTENT);
    opt.setVisibility(AQEnqueueOptions.VisibilityOption.ON_COMMIT);

    // execute the actual enqueue operation:
    conn.enqueue(queueName, opt, mesg);

这对我来说效果很好，因为我们要确保仅在提交事务后，该消息才对消费者可见。

问题-在演示中，我们创建有效负载类型RAW的队列

doUpdateDatabase(conn,
           "BEGIN "+
           "DBMS_AQADM.CREATE_QUEUE_TABLE( "+
           "   QUEUE_TABLE        =>  '"+USERNAME+".RAW_SINGLE_QUEUE_TABLE',  "+
           "   QUEUE_PAYLOAD_TYPE =>  'RAW', "+
           "   COMPATIBLE         =>  '10.0'); "+
           "END; ");
doUpdateDatabase(conn,
           "BEGIN "+
           "DBMS_AQADM.CREATE_QUEUE( "+
           "    QUEUE_NAME     =>   '"+USERNAME+".RAW_SINGLE_QUEUE', "+
           "    QUEUE_TABLE    =>   '"+USERNAME+".RAW_SINGLE_QUEUE_TABLE'); "+
           "END;  ");
doUpdateDatabase(conn,
           "BEGIN "+
           "  DBMS_AQADM.START_QUEUE('"+USERNAME+".RAW_SINGLE_QUEUE'); "+
           "END; ");

通过使用在RAW中创建的队列，我可以将消息排队到队列中，但是JMS使用者无法订阅该队列，并抛出（空指针）异常，其中使用者需要一个预期类型的​​参数。简而言之，此代码在初始化时引发空指针异常。

Properties env = new Properties();
env.load(new FileInputStream(new File("jndi.properties")));
Context ctx = new InitialContext(env);
ConnectionFactory connFactory = (ConnectionFactory)ctx.lookup(connectionFactoryName);
Connection connection = connFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
AQjmsSession queueSession = (AQjmsSession) session;
Queue queue = (Queue) ctx.lookup(queueName);
MessageConsumer receiver = queueSession.createReceiver(queue);

JNDI.properties

java.naming.factory.initial = oracle.jms.AQjmsInitialContextFactory
java.naming.security.principal = username
java.naming.security.credentials = password
db_url = jdbc:oracle:thin:@host:port:dbname

尝试在骆驼中设置使用者时，我遇到类似的异常。

<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
        <!-- this camel route will read incoming messages from Oracle -->
        <route>
            <from uri="oracleQueue:queue:RAW_SINGLE_QUEUE" />
            <to uri="WebSphereMQ:queue:myWebSphereQueue" />
        </route>
    </camelContext>

    <bean id="connectionFactoryOracleAQQueue" class="oracle.jms.AQjmsFactory" factory-method="getQueueConnectionFactory">
        <constructor-arg index="0">
            <value>oracle db URL</value>
        </constructor-arg>
        <constructor-arg index="1" type="java.util.Properties">
            <value></value>
        </constructor-arg>
    </bean>

    <bean id="oracleQueueCredentials" class="org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapter">
        <property name="targetConnectionFactory">
            <ref bean="connectionFactoryOracleAQQueue" />
        </property>
        <property name="username">
            <value>username</value>
        </property>
        <property name="password">
            <value>password</value>
        </property>
    </bean>

    <bean id="oracleQueue" class="org.apache.camel.component.jms.JmsComponent">
        <property name="connectionFactory" ref="oracleQueueCredentials" />
    </bean>

通过一些研究，我发现可能是队列有效负载类型。因此，我更改了队列表创建脚本并将JMS消息用作有效负载类型

 doUpdateDatabase(conn, "BEGIN " + "DBMS_AQADM.CREATE_QUEUE_TABLE( "
 + "   QUEUE_TABLE        =>  'RAW_SINGLE_QUEUE_TABLE',  "
 + "   QUEUE_PAYLOAD_TYPE =>  'SYS.AQ$_JMS_MESSAGE', " +
 "   COMPATIBLE         =>  '10.0'); " + "END; ");

在这种情况下，JMS使用者可以连接，但是排队代码现在失败 -ORA-25215：user_data类型和队列类型不匹配

问题是，如何才能使消息从Java生产者入队列（仅在提交时可见），并且能够与骆驼或通用JMS消费者一起使用？

约束（过滤掉网上已经存在的一些答案）-无法使用PL / SQL，spring事务，JTA。我看过一些示例，例如如何使用Java将JMS消息排队到Oracle AQ中，其中使用 SYS.AQ $ _JMS_MESSAGE类型创建队列表，但是示例生产者是JMS MessageProducer，而不是oracle指南中的示例。我不是在尝试使JMS消息（AQJmsMessage）入队，而是按照Oracle指南中的说明使用AQMessage类型，并使用visible on commit选项。

我的感觉是，如果问题仅基于有效载荷类型的不匹配，那么必须在使用者方进行某种配置以指定有效载荷类型，或者在生产者方必须进行某种配置，以便能够以JMS使用者使用的方式编写消息了解。有没有办法做到这一点？

Answer 1

我能够做到这一点-我不得不猜测Oracle API的许多部分，并从各种博客中收集提示。对于任何对此感兴趣的人来说，我的工作方式都是如此-1.我在Oracle Db上创建了一个Oracle Object 2.通过此Oracle Object，我创建了对象类型作为有效负载的队列表3.现在可以排队AQMessage类型使用STRUCT有效负载，其中包含对象数据4。而且我能够与了解ADT有效负载类型的JMS使用方出队（感谢http://blog.javaforge.net/post/30858904340/oracle-advanced-排队弹簧自定义类型）

这是使用代码的步骤-创建Oracle 对象。该对象可以具有任何主数据类型字段，例如VARCHAR，TIMESTAMP等，还可以具有BLOB，CLOB等。在这种情况下，我提供了其中的一列作为Blob，使事情变得更加复杂。

create or replace type aq_event_obj as object
(
  id       varchar2(100),
  payload  BLOB
);
commit;

现在创建队列表。表的有效负载类型是oracle对象。

private void setup(Connection conn) throws SQLException {
    doUpdateDatabase(conn, "BEGIN " + "DBMS_AQADM.CREATE_QUEUE_TABLE( "
            + "   QUEUE_TABLE        =>  'OBJ_SINGLE_QUEUE_TABLE',  " + "   QUEUE_PAYLOAD_TYPE =>  'AQ_EVENT_OBJ', "
            + "   COMPATIBLE         =>  '10.0'); " + "END; ");
    doUpdateDatabase(conn, "BEGIN " + "DBMS_AQADM.CREATE_QUEUE( " + "    QUEUE_NAME     =>   'OBJ_SINGLE_QUEUE', "
            + "    QUEUE_TABLE    =>   'OBJ_SINGLE_QUEUE_TABLE'); " + "END;  ");
    doUpdateDatabase(conn, "BEGIN " + "  DBMS_AQADM.START_QUEUE('OBJ_SINGLE_QUEUE'); " + "END; ");
} 

您现在可以使用对象的struct实例将Java中的AQMessage类型排队

public void enqueueMessage(OracleConnection conn, String correlationId, byte[] payloadData) throws Exception {
    // First create the message properties:
    AQMessageProperties aqMessageProperties = AQFactory.createAQMessageProperties();
    aqMessageProperties.setCorrelation(correlationId);
    aqMessageProperties.setExceptionQueue(EXCEPTION_QUEUE_NAME);

    // Specify an agent as the sender:
    AQAgent aqAgent = AQFactory.createAQAgent();
    aqAgent.setName(SENDER_NAME);
    aqAgent.setAddress(QUEUE_NAME);
    aqMessageProperties.setSender(aqAgent);

    // Create the payload
    StructDescriptor structDescriptor = StructDescriptor.createDescriptor(EVENT_OBJECT, conn);
    Map<String, Object> payloadMap = new HashMap<String, Object>();
    payloadMap.put("ID", correlationId);
    payloadMap.put("PAYLOAD", new OracleAQBLOBUtil().createBlob(conn, payloadData));
    STRUCT struct = new STRUCT(structDescriptor, conn, payloadMap);

    // Create the actual AQMessage instance:
    AQMessage aqMessage = AQFactory.createAQMessage(aqMessageProperties);
    aqMessage.setPayload(struct);

    AQEnqueueOptions opt = new AQEnqueueOptions();
    opt.setDeliveryMode(AQEnqueueOptions.DeliveryMode.PERSISTENT);
    opt.setVisibility(AQEnqueueOptions.VisibilityOption.ON_COMMIT);

    // execute the actual enqueue operation:
    conn.enqueue(QUEUE_NAME, opt, aqMessage);
}

Blob字段需要特殊处理

public class OracleAQBLOBUtil {

    public BLOB createBlob(OracleConnection conn, byte[] payload) throws Exception {
        BLOB blob = BLOB.createTemporary(conn, false, BLOB.DURATION_SESSION);
        OutputStream outputStream = blob.setBinaryStream(1L);
        InputStream inputStream = new ByteArrayInputStream(payload);
        try {
            byte[] buffer = new byte[blob.getBufferSize()];
            int bytesRead = 0;
            while ((bytesRead = inputStream.read(buffer)) != -1) {
                outputStream.write(buffer, 0, bytesRead);
            }
            return blob;
        }
        finally {
            outputStream.close();
            inputStream.close();
        }
    }

    public byte[] saveOutputStream(BLOB blob) throws Exception {
        InputStream inputStream = blob.getBinaryStream();
        int counter;
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        while ((counter = inputStream.read()) > -1) {
            byteArrayOutputStream.write(counter);
        }
        byteArrayOutputStream.close();
        return byteArrayOutputStream.toByteArray();
    }

}

对于使用者，您需要提供一个ORADataFactory实例，以使使用者了解有效负载类型（您的自定义对象）。

AQjmsSession queueSession = (AQjmsSession) session;
Queue queue = (Queue) ctx.lookup(queueName);
MessageConsumer receiver = queueSession.createReceiver(queue, new OracleAQObjORADataFactory());

OracleAQObjORADataFactory的代码在哪里

import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.SQLException;

import oracle.jdbc.OracleTypes;
import oracle.jpub.runtime.MutableStruct;
import oracle.sql.BLOB;
import oracle.sql.Datum;
import oracle.sql.ORAData;
import oracle.sql.ORADataFactory;
import oracle.sql.STRUCT;

public class OracleAQObjORADataFactory  implements ORAData, ORADataFactory {

    public static final String EVENT_OBJECT = "SYSTEM.AQ_EVENT_OBJ";
    public static final int _SQL_TYPECODE = OracleTypes.STRUCT;

    protected MutableStruct _struct;

    protected static int[] _sqlType = { java.sql.Types.VARCHAR, java.sql.Types.VARBINARY };
    protected static ORADataFactory[] _factory = new ORADataFactory[2];
    protected static final OracleAQObjORADataFactory  _AqEventObjFactory = new OracleAQObjORADataFactory ();

    public static ORADataFactory getORADataFactory() {
        return _AqEventObjFactory;
    }

    /* constructors */
    protected void _init_struct(boolean init) {
        if (init)
            _struct = new MutableStruct(new Object[2], _sqlType, _factory);
    }

    public OracleAQObjORADataFactory () {
        _init_struct(true);
    }

    public OracleAQObjORADataFactory (String id, byte[] payload) throws SQLException {
        _init_struct(true);
        setId(id);
        setPayload(payload);
    }

    /* ORAData interface */
    public Datum toDatum(Connection c) throws SQLException {
        return _struct.toDatum(c, EVENT_OBJECT);
    }

    /* ORADataFactory interface */
    public ORAData create(Datum d, int sqlType) throws SQLException {
        return create(null, d, sqlType);
    }

    protected ORAData create(OracleAQObjORADataFactory  o, Datum d, int sqlType) throws SQLException {
        if (d == null)
            return null;
        if (o == null)
            o = new OracleAQObjORADataFactory ();
        o._struct = new MutableStruct((STRUCT) d, _sqlType, _factory);
        return o;
    }

    public String getId() throws SQLException {
        return (String) _struct.getAttribute(0);
    }

    public void setId(String id) throws SQLException {
        _struct.setAttribute(0, id);
    }

    public byte[] getPayload() throws SQLException {
        BLOB blob = (BLOB) _struct.getAttribute(1);
        InputStream inputStream = blob.getBinaryStream();
        return getBytes(inputStream);
    }

    public byte[] getBytes(InputStream body) {
        int c;
        try {
            ByteArrayOutputStream f = new ByteArrayOutputStream();
            while ((c = body.read()) > -1) {
                f.write(c);
            }
            f.close();
            byte[] result = f.toByteArray();
            return result;
        }
        catch (Exception e) {
            System.err.println("Exception: " + e.getMessage());
            e.printStackTrace();
            return null;
        }
    }

    public void setPayload(byte[] payload) throws SQLException {
        _struct.setAttribute(1, payload);
    }

}

在这种情况下，您可能正在项目中使用Camel或Spring-1.如果您使用的是Camel 2.10.2或更高版本，则可以使用自定义消息列表容器（CAMEL-5676）创建JMS使用者。2.如果如果您使用的是以前的版本，则可能无法使用端点方式（我无法弄清楚），但是可以使用JMS请求侦听器

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jms="http://www.springframework.org/schema/jms"
    xmlns:p="http://www.springframework.org/schema/p"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
                        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
                        http://www.springframework.org/schema/jms
                        http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">

    <!-- this is just an example, you can also use a datasource as the ctor arg -->
    <bean id="connectionFactoryOracleAQQueue" class="oracle.jms.AQjmsFactory" factory-method="getQueueConnectionFactory">
        <constructor-arg index="0">
            <value>jdbc:oracle:thin:@blrub442:1522:UB23</value>
        </constructor-arg>
        <constructor-arg index="1" type="java.util.Properties">
            <value></value>
        </constructor-arg>
    </bean>

    <bean id="oracleQueueCredentials" class="org.springframework.jms.connection.UserCredentialsConnectionFactoryAdapter">
        <property name="targetConnectionFactory">
            <ref bean="connectionFactoryOracleAQQueue" />
        </property>
        <property name="username">
            <value>system</value>
        </property>
        <property name="password">
            <value>oracle</value>
        </property>
    </bean>

    <!-- Definitions for JMS Listener classes that we have created -->
    <bean id="aqMessageListener" class="com.misys.test.JmsRequestListener" />

    <bean id="aqEventQueue" class="com.misys.test.OracleAqQueueFactoryBean">
        <property name="connectionFactory" ref="oracleQueueCredentials" />
        <property name="oracleQueueName" value="BOZ_SINGLE_QUEUE" />
    </bean>

    <!-- The Spring DefaultMessageListenerContainer configuration. This bean is automatically loaded when the JMS application context is started -->
    <bean id="jmsContainer" class="com.misys.test.AQMessageListenerContainer" scope="singleton">
        <property name="connectionFactory" ref="oracleQueueCredentials" />
        <property name="destination" ref="aqEventQueue" />
        <property name="messageListener" ref="aqMessageListener" />
        <property name="sessionTransacted" value="false" />
    </bean>

</beans>

自定义消息侦听器容器

public class AQMessageListenerContainer extends DefaultMessageListenerContainer {

    @Override
    protected MessageConsumer createConsumer(Session session, Destination destination) throws JMSException {
        return ((AQjmsSession) session).createConsumer(destination, getMessageSelector(),
                OracleAQObjORADataFactory.getORADataFactory(), null, isPubSubNoLocal());
    }
}

和请求侦听器的onMessage方法

public void onMessage(Message msg) {
    try {
        AQjmsAdtMessage aQjmsAdtMessage = (AQjmsAdtMessage) msg;
        OracleAQObjORADataFactory obj = (OracleAQObjORADataFactory) aQjmsAdtMessage.getAdtPayload();

        System.out.println("Datetime: " + obj.getId());
        System.out.println("Payload: " + new String(obj.getPayload(), Charset.forName("UTF-8")));
    }
    catch (Exception jmsException) {
        if (logger.isErrorEnabled()) {
            logger.error(jmsException.getLocalizedMessage());
        }
    }
}