この章では、Oracle Streams Advanced Queuing(AQ)に対するJava Message Service(JMS)操作インタフェース(共有インタフェース)を説明します。
内容は次のとおりです。
この項では、JMS操作のためのOracle Streams AQ共有インタフェースを説明します。
内容は次のとおりです。
public void start() throws JMSException
AQjmsConnection.start()
は、メッセージ受信用のJMSコネクションを起動します。
public oracle.jms.AQjmsConnection getJmsConnection() throws JMSException
AQjmsSession.getJmsConnection()
は、セッションからJMSコネクションを取得します。
public void commit() throws JMSException
AQjmsSession.commit()
は、セッションで実行されているすべてのJMSおよびSQL操作をコミットします。
public void rollback() throws JMSException
AQjmsSession.rollback()
は、セッションで実行されているすべてのJMSおよびSQL操作を終了します。
public java.sql.Connection getDBConnection() throws JMSException
AQjmsSession.getDBConnection()
は、JMSセッションから基礎となるJDBCコネクションを取得します。JDBCコネクションは、JMS操作が実行される同一のトランザクションの一部としてSQL操作を実行するために使用される場合があります。
public oracle.jdbc.pool.OracleOCIConnectionPool getOCIConnectionPool()
AQjmsConnection.getOCIConnectionPool()
は、JMSコネクションから基礎となるOracleOCIConnectionPool
を取得します。OracleOCIConnectionPool
インスタンスの設定は、コネクションの使用状況(ユーザーが既存のコネクションを使用して確立するセッション数など)に応じてユーザーがチューニングできます。ただし、JMSコネクションが使用しているOracleOCIConnectionPool
インスタンスはクローズしないでください。
public javax.jms.BytesMessage createBytesMessage() throws JMSException
AQjmsSession.createBytesMessage()
は、Bytesメッセージを作成します。このメソッドを使用できるのは、宛先キュー/トピックを含むキュー表がペイロード型SYS.AQ$_JMS_BYTE_MESSAGE
またはAQ$_JMS_MESSAGE
で作成された場合のみです。
public javax.jms.MapMessage createMapMessage() throws JMSException
AQjmsSession.createMapMessage()
は、Mapメッセージを作成します。このメソッドを使用できるのは、宛先キュー/トピックを含むキュー表がペイロード型SYS.AQ$_JMS_MAP_MESSAGE
またはAQ$_JMS_MESSAGE
で作成された場合のみです。
public javax.jms.StreamMessage createStreamMessage() throws JMSException
AQjmsSession.createStreamMessage()
は、StreamMessageを作成します。このメソッドを使用できるのは、宛先キュー/トピックを含むキュー表がペイロード型SYS.AQ$_JMS_STREAM_MESSAGE
またはAQ$_JMS_MESSAGE
で作成された場合のみです。
public javax.jms.ObjectMessage createObjectMessage(java.io.Serializable object) throws JMSException
AQjmsSession.createObjectMessage()
は、Objectメッセージを作成します。このメソッドを使用できるのは、宛先キュー/トピックを含むキュー表がペイロード型SYS.AQ$_JMS_OBJECT_MESSAGE
またはAQ$_JMS_MESSAGE
で作成された場合のみです。
public javax.jms.TextMessage createTextMessage() throws JMSException
AQjmsSession.createTextMessage()
は、Textメッセージを作成します。このメソッドを使用できるのは、宛先キュー/トピックを含むキュー表がペイロード型SYS.AQ$_JMS_TEXT_MESSAGE
またはAQ$_JMS_MESSAGE
で作成された場合のみです。
public javax.jms.Message createMessage() throws JMSException
AQjmsSession.createMessage()
は、JMSメッセージを作成します。AQ$_JMS_MESSAGE
メッセージを使用して、様々な型のメッセージを構成できます。メッセージ型は、次のいずれかである必要があります。
DBMS_AQ.JMS_TEXT_MESSAGE
DBMS_AQ.JMS_OBJECT_MESSAGE
DBMS_AQ.JMS_MAP_MESSAGE
DBMS_AQ.JMS_BYTES_MESSAGE
DBMS_AQ.JMS_STREAM_MESSAGE
このユーザー定義型を使用して、ヘッダーのみのJMSメッセージを作成することもできます。
public oracle.jms.AdtMessage createAdtMessage() throws JMSException
AQjmsSession.createAdtMessage()
はAdtMessage
を作成します。このメソッドを使用できるのは、キュー/トピックを含むキュー表がOracle ADTペイロード型で作成された場合のみです。AdtMessage
は、CustomDatum
インタフェースを実装するオブジェクトで移入される必要があります。このオブジェクトは、キュー/トピックのペイロードとして定義されている、SQLのユーザー定義型Javaマッピングである必要があります。SQLのユーザー定義型に対応するJavaクラスは、JPublisherツールを使用して生成できます。
JMSで始まるプロパティ名は、プロバイダ固有です。ユーザー定義のプロパティは、JMSで始めることができません。
次のプロバイダのプロパティは、Text、Stream、Object、BytesまたはMapMessageを使用するクライアントが設定できます。
JMSXAppID(String)
JMSXGroupID(String)
JMSXGroupSeq(Int)
JMS_OracleExcpQ(String)
このメッセージ・プロパティでは例外キューを指定します。
JMS_OracleDelay(Int)
このメッセージ・プロパティではメッセージの遅延秒数を指定します。
次のプロパティは、AdtMessage
に対して設定できます。
JMS_OracleExcpQ(String)
このメッセージ・プロパティでは、例外キューをschema
.queue_name
として指定します。
JMS_OracleDelay(Int)
このメッセージ・プロパティではメッセージの遅延秒数を指定します。
内容は次のとおりです。
public void setBooleanProperty(java.lang.String name, boolean value) throws JMSException
AQjmsMessage.setBooleanProperty()
は、メッセージ・プロパティをBooleanとして指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Booleanプロパティの名前 |
value
|
メッセージに設定するBooleanプロパティの値 |
public void setStringProperty(java.lang.String name, java.lang.String value) throws JMSException
AQjmsMessage.setStringProperty()
は、メッセージ・プロパティをStringとして指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Stringプロパティの名前 |
value
|
メッセージに設定するStringプロパティの値 |
public void setIntProperty(java.lang.String name, int value) throws JMSException
AQjmsMessage.setIntProperty()
は、メッセージ・プロパティをIntegerとして指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Integerプロパティの名前 |
value
|
メッセージに設定するIntegerプロパティの値 |
public void setDoubleProperty(java.lang.String name, double value) throws JMSException
AQjmsMessage.setDoubleProperty()
は、メッセージ・プロパティをDoubleとして指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Doubleプロパティの名前 |
value
|
メッセージに設定するDoubleプロパティの値 |
public void setFloatProperty(java.lang.String name, float value) throws JMSException
AQjmsMessage.setFloatProperty()
は、メッセージ・プロパティをFloatとして指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Floatプロパティの名前 |
value
|
メッセージに設定するFloatプロパティの値 |
public void setByteProperty(java.lang.String name, byte value) throws JMSException
AQjmsMessage.setByteProperty()
は、メッセージ・プロパティをByteとして指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Byteプロパティの名前 |
value
|
メッセージに設定するByteプロパティの値 |
public void setLongProperty(java.lang.String name, long value) throws JMSException
AQjmsMessage.setLongProperty()
は、メッセージ・プロパティをLongとして指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Longプロパティの名前 |
value
|
メッセージに設定するLongプロパティの値 |
public void setShortProperty(java.lang.String name, short value) throws JMSException
AQjmsMessage.setShortProperty()
は、メッセージ・プロパティをShortとして指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Shortプロパティの名前 |
value
|
メッセージに設定するShortプロパティの値 |
public void setObjectProperty(java.lang.String name, java.lang.Object value) throws JMSException
AQjmsMessage.setObjectProperty()
は、メッセージ・プロパティをObjectとして指定します。オブジェクトとしてのプリミティブの値(Boolean、Byte、Short、Integer、Long、Float、DoubleおよびString)のみがサポートされています。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
JavaのObjectプロパティの名前 |
value
|
メッセージに設定するJavaのObjectプロパティの値 |
public void setTimeToLive(long timeToLive) throws JMSException
このメソッドは、MessageProducer
が送信するすべてのメッセージに対するデフォルトのTimeToLive
を設定します。TimeToLiveはメッセージ遅延が有効になった後に計算されます。このメソッドのパラメータは次のとおりです。
パラメータ | 説明 |
---|---|
timeToLive
|
ミリ秒で指定されるメッセージのTime-To-Live(ゼロは無制限) |
public void setPriority(int priority) throws JMSException
このメソッドは、MessageProducer
が送信するすべてのメッセージに対するデフォルトのPriority
を設定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
priority
|
このMessageProducerのメッセージの優先順位。デフォルトは4です。 |
優先順位には、どの整数値でも指定できます。数値が小さいほど、優先順位が高いことを表します。優先順位がsend()
操作時に明示的に指定されている場合は、このメソッドで設定されたデフォルト値をオーバーライドします。
public void createAQAgent(java.lang.String agent_name, boolean enable_http, throws JMSException
このメソッドはAQjmsAgent
を作成します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
agent_name
|
AQエージェントの名前。 |
enable_http
|
TRUEに設定すると、このエージェントはHTTP経由でAQにアクセスできます。 |
タイムアウトを指定するか、待機なしで、メッセージを同期受信できます。また、変換を使用してメッセージを受信することもできます。
public javax.jms.Message receive(long timeout) throws JMSException
このメソッドは、タイムアウトを指定し、メッセージ・コンシューマを使用してメッセージを受信します。
パラメータ | 説明 |
---|---|
timeout
|
タイムアウト値(ミリ秒単位) |
例15-6 タイムアウト指定でのメッセージ・コンシューマの使用
TopicConnectionFactory tc_fact = null; TopicConnection t_conn = null; TopicSession t_sess = null; TopicSession jms_sess; Topic shipped_orders; int myport = 5521; /* create connection and session */ tc_fact = AQjmsFactory.getTopicConnectionFactory( "MYHOSTNAME", "MYSID", myport, "oci8"); t_conn = tc_fact.createTopicConnection("jmstopic", "jmstopic"); jms_sess = t_conn.createTopicSession(true, Session.CLIENT_ACKNOWLEDGE); shipped_orders = ((AQjmsSession )jms_sess).getTopic( "WS", "Shipped_Orders_Topic"); /* create a subscriber, specifying the correct CustomDatumFactory and selector */ subscriber1 = jms_sess.createDurableSubscriber( shipped_orders, 'WesternShipping', " priority > 1 and tab.user_data.region like 'WESTERN %'", false, AQjmsAgent.getFactory()); /* receive, blocking for 30 seconds if there were no messages */ Message = subscriber.receive(30000);
例15-7 JMS: メッセージが届くまでブロック
public BolOrder get_new_order1(QueueSession jms_session) { Queue queue; QueueReceiver qrec; ObjectMessage obj_message; BolCustomer customer; BolOrder new_order = null; String state; try { /* get a handle to the new_orders queue */ queue = ((AQjmsSession) jms_session).getQueue("OE", "OE_neworders_que"); qrec = jms_session.createReceiver(queue); /* wait for a message to show up in the queue */ obj_message = (ObjectMessage)qrec.receive(); new_order = (BolOrder)obj_message.getObject(); customer = new_order.getCustomer(); state = customer.getState(); System.out.println("Order: for customer " + customer.getName()); } catch (JMSException ex) { System.out.println("Exception: " + ex); } return new_order; }
public javax.jms.Message receiveNoWait() throws JMSException
このメソッドは、待機なしで、メッセージ・コンシューマを使用してメッセージを受信します。
例15-8 JMS: メッセージを非ブロック
public BolOrder poll_new_order3(QueueSession jms_session) { Queue queue; QueueReceiver qrec; ObjectMessage obj_message; BolCustomer customer; BolOrder new_order = null; String state; try { /* get a handle to the new_orders queue */ queue = ((AQjmsSession) jms_session).getQueue("OE", "OE_neworders_que"); qrec = jms_session.createReceiver(queue); /* check for a message to show in the queue */ obj_message = (ObjectMessage)qrec.receiveNoWait(); new_order = (BolOrder)obj_message.getObject(); customer = new_order.getCustomer(); state = customer.getState(); System.out.println("Order: for customer " + customer.getName()); } catch (JMSException ex) { System.out.println("Exception: " + ex); } return new_order; }
キューまたはトピックからメッセージを受信するときに変換を適用できます。メッセージを変換してから、JMSアプリケーションに戻します。
変換は、AqjmsQueueReceiver
、AQjmsTopicSubscriber
またはAQjmsTopicReceiver
のsetTransformation()
インタフェースを使用して指定できます。
例15-9 JMS: 変換による宛先からのメッセージの受信
西部地域向け出荷処理アプリケーションがOE_bookedorders_topicからメッセージを取り出すと仮定します。変換OE2WS
を指定して、メッセージをOracleオブジェクト型WS_order
として取り出します。WSOrder JavaクラスがJPublisherによって生成され、Oracleオブジェクト型WS.WS_order
にマップされると仮定します。
public AQjmsAdtMessage retrieve_bookedorders(TopicSession jms_session) AQjmsTopicReceiver receiver; Topic topic; Message msg = null; try { /* get a handle to the OE_bookedorders_topic */ topic = ((AQjmsSession)jms_session).getTopic("OE", "OE_bookedorders_topic"); /* Create a receiver for WShip */ receiver = ((AQjmsSession)jms_session).createTopicReceiver( topic, "WShip, null, WSOrder.getFactory()); /* set the transformation in the publisher */ receiver.setTransformation("OE2WS"); msg = receiver.receive(10); } catch (JMSException ex) { System.out.println("Exception :", ex); } return (AQjmsAdtMessage)msg; }
public void setNavigationMode(int mode) throws JMSException
このメソッドは、メッセージの受信に対してナビゲーション・モードを指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
mode
|
ナビゲーション・モードの新規値 |
例15-10 メッセージの受信に対するナビゲーション・モードの指定
TopicConnectionFactory tc_fact = null; TopicConnection t_conn = null; TopicSession t_sess = null; TopicSession jms_sess; Topic shipped_orders; int myport = 5521; /* create connection and session */ tc_fact = AQjmsFactory.getTopicConnectionFactory( "MYHOSTNAME", "MYSID", myport, "oci8"); t_conn = tc_fact.createTopicConnection("jmstopic", "jmstopic"); jms_sess = t_conn.createTopicSession(true, Session.CLIENT_ACKNOWLEDGE); shipped_orders = ((AQjmsSession )jms_sess).getTopic("WS", "Shipped_Orders_Topic"); /* create a subscriber, specifying the correct CustomDatumFactory and selector */ subscriber1 = jms_sess.createDurableSubscriber( shipped_orders, 'WesternShipping', "priority > 1 and tab.user_data.region like 'WESTERN %'", false, AQjmsAgent.getFactory()); subscriber1.setNavigationMode(AQjmsConstants.NAVIGATION_FIRST_MESSAGE); /* get message for the subscriber, returning immediately if there was nomessage */ Message = subscriber.receive();
メッセージを非同期に受信するには、次の2つの方法があります。
public void setMessageListener(javax.jms.MessageListener myListener) throws JMSException
このメソッドは、メッセージ・コンシューマに対してメッセージ・リスナーを指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
myListener
|
コンシューマに対してメッセージ・リスナーを設定します。 |
例15-11 メッセージ・コンシューマに対するメッセージ・リスナーの指定
TopicConnectionFactory tc_fact = null; TopicConnection t_conn = null; TopicSession t_sess = null; TopicSession jms_sess; Topic shipped_orders; int myport = 5521; MessageListener mLis = null; /* create connection and session */ tc_fact = AQjmsFactory.getTopicConnectionFactory( "MYHOSTNAME", "MYSID", myport, "oci8"); t_conn = tc_fact.createTopicConnection("jmstopic", "jmstopic"); jms_sess = t_conn.createTopicSession(true, Session.CLIENT_ACKNOWLEDGE); shipped_orders = ((AQjmsSession )jms_sess).getTopic( "WS", "Shipped_Orders_Topic"); /* create a subscriber, specifying the correct CustomDatumFactory and selector */ subscriber1 = jms_sess.createDurableSubscriber( shipped_orders, 'WesternShipping', "priority > 1 and tab.user_data.region like 'WESTERN %'", false, AQjmsAgent.getFactory()); mLis = new myListener(jms_sess, "foo"); /* get message for the subscriber, returning immediately if there was nomessage */ subscriber.setMessageListener(mLis); The definition of the myListener class import oracle.AQ.*; import oracle.jms.*; import javax.jms.*; import java.lang.*; import java.util.*; public class myListener implements MessageListener { TopicSession mySess; String myName; /* constructor */ myListener(TopicSession t_sess, String t_name) { mySess = t_sess; myName = t_name; } public onMessage(Message m) { System.out.println("Retrieved message with correlation: " || m.getJMSCorrelationID()); try{ /* commit the dequeue */ mySession.commit(); } catch (java.sql.SQLException e) {System.out.println("SQL Exception on commit"); } } }
内容は次のとおりです。
public java.lang.String getJMSCorrelationID() throws JMSException
AQjmsMessage.getJMSCorrelationID()
は、メッセージの相関識別子を取得します。
内容は次のとおりです。
public boolean getBooleanProperty(java.lang.String name) throws JMSException
AQjmsMessage.getBooleanProperty()
は、メッセージ・プロパティをBooleanとして取得します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Booleanプロパティの名前 |
public java.lang.String getStringProperty(java.lang.String name) throws JMSException
AQjmsMessage.getStringProperty()
は、メッセージ・プロパティをStringとして取得します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Stringプロパティの名前 |
public int getIntProperty(java.lang.String name) throws JMSException
AQjmsMessage.getIntProperty()
は、メッセージ・プロパティをIntegerとして取得します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Integerプロパティの名前 |
public double getDoubleProperty(java.lang.String name) throws JMSException
AQjmsMessage.getDoubleProperty()
は、メッセージ・プロパティをDoubleとして取得します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Doubleプロパティの名前 |
public float getFloatProperty(java.lang.String name) throws JMSException
AQjmsMessage.getFloatProperty()
は、メッセージ・プロパティをFloatとして取得します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Floatプロパティの名前 |
public byte getByteProperty(java.lang.String name) throws JMSException
AQjmsMessage.getByteProperty()
は、メッセージ・プロパティをByteとして取得します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Byteプロパティの名前 |
public long getLongProperty(java.lang.String name) throws JMSException
AQjmsMessage.getLongProperty()
は、メッセージ・プロパティをLongとして取得します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Longプロパティの名前 |
public short getShortProperty(java.lang.String name) throws JMSException
AQjmsMessage.getShortProperty()
は、メッセージ・プロパティをShortとして取得します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
name
|
Shortプロパティの名前 |
内容は次のとおりです。
public void close() throws JMSException
AQjmsProducer.close()
はMessageProducer
をクローズします。
public void close() throws JMSException
AQjmsConsumer.close()
は、メッセージ・コンシューマをクローズします。
public void close() throws JMSException
AQjmsSession.close()
はJMSセッションをクローズします。
内容は次のとおりです。
public java.lang.String getErrorCode()
AQjmsException.getErrorCode()
は、JMS例外のエラー・コードを取得します。
public int getErrorNumber()
AQjmsException.getErrorNumber()
は、JMS例外のエラー番号を取得します。
注意: このメソッドは今後のリリースでは使用されなくなります。かわりに、getErrorCode() を使用してください。
|
public java.lang.String getLinkString()
AQjmsException.getLinkString()
は、JMS例外にリンクされた例外を取得します。一般に、この例外にはデータベースで発行されるSQL例外が含まれます。
public void printStackTrace(java.io.PrintStream s)
AQjmsException.printStackTrace()
は、JMS例外のスタック・トレースを出力します。
public void setExceptionListener(javax.jms.ExceptionListener listener) throws JMSException
AQjmsConnection.setExceptionListener()
は、コネクション用の例外リスナーを指定します。パラメータは次のとおりです。
パラメータ | 説明 |
---|---|
listener
|
例外リスナー |
例外リスナーの登録が完了している場合は、コネクションに重大な問題が検出されると通知されます。これは、リスナーのonException()
メソッドをコールして、問題を説明するJMS例外を渡すことによって実行されます。これによって、問題がJMSクライアントに非同期に通知されます。メッセージを処理するのみのコネクションもあるため、コネクションが失敗したことを知るための他の方法がありません。