JMSConcepts

• 信頼性のある、非同期で、緩い結合のメッセージングＡＰＩ

---

Overviewof the JMS API

What Is Messaging

• ソフトウェアコンポーネントやソフトウェアアプリケーション間のPtoP通信手段
  • クライアントはメッセージの送受信をおこなう。
  • クライアントはメッセージエージェントを利用
• メッセージングは緩い結合
  • 送信者と受信者は同時に通信しなくていい
  • 送信者と受信者はお互いのことをしらなくていい。
    • メッセージの形式と方向だけ知っていればいい

What Is the JMS API?

• アプリケーションがメッセージを作成、送信、受信、読み込みするためのAPI
• 緩い結合を可能にする
• 非同期通信を可能にする
• 信頼性
  • メッセージが１度だけ届くことを保証

When Can You Use the JMS API?

• コンポーネントが他のコンポーネントのIFなどの情報に依存しないようにしたい
• すべてのコンポーネントが起動して同時に実行されていなくても、アプリケーションを実行したい
• コンポーネントが別コンポーネントに要求を送った後、直ちに後続の処理を行いたい

How Does the JMS API Work with the Java EE Platform?

• JavaEEにおけるJMSの特徴
  • アプリケーションクライアント、EJBコンポーネント、WEBコンポーネントはJMSメッセージを送信、同期的に受信できる
    • アプリケーションクライアントは非同期でのメッセージ受信も可能
  • メッセージ駆動ビーンを使うと、メッセージを非同期で集約できる
• JMSを使って、JavaEEコンポーネントとレガシーシステムをメッセージング連携できる
  • メッセージ駆動ビーンを作成することで、ビジネスイベントに対する振る舞いを追加できる
  • JavaEEプラットフォームはJMS APIを強化している
    • 分散トランザクション
    • 平行処理
  • JavaEEコネクタアーキテクチャーを使って、JMSプロバイダをサーバにプラグインできる

---

Basic JMS API Concepts

JMS API Architecture

• JMSアプリケーションは以下の構成要素からなる
  • JMSプロバイダ
  • JMSクライアント
  • メッセージ
  • 管理オブジェクト

Messaging Domains

Point-to-Point Messaging Domain

Publish/Subscribe Messaging Domain

Programming with the CommonInterfaces

• JMS API v1.1ではPTPとPub/Subドメインで、メッセージの送受信を行うコードは同じにできる
  • キューを使うかトピックを使うかで全体の動きは違うが、メッセージを処理するコードは同じ

Message Consumption

• JMSでのメッセージの消費のされ方は２通り
  • 同期
    • サブスクライバまたはレシーバはメッセージをDestinationからreceive()によって取得する
    • receiveメソッドはメッセージが到着するまでブロックすることができる
    • receiveメソッドはメッセージが決められた時間内に到着しなければタイムアウトできる
  • 非同期
    • コンシューマをリスナーに登録
    • メッセージがDestinationに来たら、JMSプロバイダはlistener.onMessageを呼んでメッセージを届ける

---

The JMS API Programming Model

• JMSアプリケーションの構成要素
  • 管理オブジェクト
    • コネクションファクトリ
    • Destination
  • コネクション
  • セッション
  • メッセージプロデューサ
  • メッセージコンシューマ
  • メッセージ

JMS Administered Objects

• Destinationとコネクションファクトリの管理はJMSプロバイダによって大きく異なる
• JMSクライアントはこれらのオブジェクトにポータブルなIFを利用してアクセスする
  • 管理者はJNDIに管理オブジェクトを設定する
  • JMSクライアントはリソースインジェクションをつかってこれらのオブジェクトにアクセス
• glassfishでは以下の方法で設定できる
  • asadmin create-jms-resourceコマンド
  • 管理コンソールからコネクタリソースとして
  • glassfish-resources.xml

JMS Connection Factories

• ConnectionFactory/QueueConnectionFactory/TopicConnectionFactory
• プロバイダに対するコネクションを生成するためにJMSクライアントが利用する
• 管理者が設定するコネクション設定パラメータを隠蔽
• コネクションファクトリはリソースインジェクションして使う

  @Resource(lookup = "jms/ConnectionFactory")
  private static ConnectionFactory connectionFactory;
  

• JavaEEでは、JMS管理オブジェクトはjmsネーミングサブコンテキスト下

JMS Destinations

• クライアントがメッセージ送信対象やメッセージ消費ソースを特定するために利用
  • PTPメッセージングドメインではキュー
  • Pub/Subメッセージングドメインではトピック
  • JMSでは複数のキューとトピック、その両方をつかうことができる
• glassfishサーバでは、デスティネーションのJNDI名を特定するJMSデスティネーションリソースを作成する
  • デスティネーションリソースは物理的なデスティネーションを参照
    • 明示的に作成
    • 暗黙的にアプリケーションサーバが作成・削除
• デスティネーションもリソースインジェクションして利用する
• トピックにもキューにも対応できるクライアントプログラムを作成するには、Destinationオブジェクトとしてインジェクションする

JMS Connections

• JMSプロバイダとの仮想的なコネクション
  • クライアントとプロバイダデーモンとのTCP/IPソケットコネクション
• コネクションを１つ以上のセッションの生成に利用する
• Connectionインターフェース実装のオブジェクト

  connectionFactory.createConnection();
  

• 作成したコネクションはクローズする
  • コネクションのクローズによって、セッションおよび、セッションのメッセージプロデューサ／コンシューマもクローズされる

    connection.close();
    

• メッセージが消費される前に、connection.startメソッドをよばないとだめ

JMS Sessions

• メッセージ生成、消費のためのシングルスレッドコンテキスト
• セッションを使って以下を生成
• 一時的にメッセージ配信を止める場合は、connection.stopメソッド
  • メッセージプロデューサ
  • メッセージコンシューマ
  • メッセージ
  • キューブラウザ
  • 一時キュー、トピック
• セッションはメッセージリスナの実行をシリアライズする
• セッションは送受信のセットをアトミックにグルーピングするためのトランザクションコンテキストを提供
• Sessionインターフェース実装オブジェクト

  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  

JMS Message Producers

• デスティネーションにメッセージ送信するためのオブジェクト
• MessageProducer実装オブジェクト

  MessageProducer producer = session.createProducer(dest|queue|topic);
  

• producer.sendでメッセージ送信
• session.createProducerにnullを渡すことで、メッセージ送信までデスティネーションを特定しないプロデューサを作成できる

  anon_prod.send(dest, message);
  

JMS Message Consumers

• デスティネーションに送信されたメッセージを受信するためのオブジェクト
• JMSクライアントがデスティネーションに対する興味をコンシューマとして登録
• MessageConsumer実装オブジェクト

  MessageConsumer consumer = session.createConsumer(dest|queue|topic);
  

  • createDurableSubscriberで永続的なトピックサブスクライバを生成できる
• consumer.close()でメッセージコンシューマを非アクティブにできる
  • connection.start()を呼ぶまでは、メッセージ配信が行われない
• メッセージを同期的に受信するにはconsumer.receive()

JMS Message Listeners

• メッセージに対する非同期イベントハンドラ
• MessageListener実装オブジェクト
  • onMessage()：メッセージが到着したときの振る舞い
• MessageConsumerにsetMessageListenerでリスナを登録する
  • リスナを登録した後に、session.startを呼ぶ
• メッセージリスナは特定のデスティネーションの型に依存しない
• メッセージリスナは特定のメッセージの型とフォーマットに依存する
• onMessageメソッドですべての例外をハンドリングする
  • チェック例外をthrowしない
  • 実行時例外はプログラムエラー
• メッセージコンシューマを生成したセッションはすべてのメッセージリスナの実行をシリアライズする
  • ある時点では、メッセージリスナの一つのみが実行される

JMS Message Selectors

• コンシューマが受け取るメッセージをフィルタすることができる
• JMSプロバイダにメッセージフィルタリング処理を割り当てる
• メッセージセレクタはSQL92の条件句の表現文字列
  • createConsumer, createDurableSubscriberの引数で指定

JMS Messages

• JMSメッセージはシンプルで柔軟性のある基本フォーマットを持つ
  • JMSアプリケーションでなくても連携できる

Message Headers

JMSDestination	send or publish method
JMSDeliveryMode	send or publish method
JMSExpiration	send or publish method
JMSPriority	send or publish method
JMSMessageID	send or publish method
JMSTimestamp	send or publish method
JMSCorrelationID	Client
JMSReplyTo	Client
JMSType	Client
JMSRedelivered	JMS provider

Message Properties

• ヘッダフィールド以外に、ユーザがプロパティを定義できる
  • 他システムとの連携
  • メッセージセレクタで使用

Message Bodies

• 5つのメッセージボディフォーマット（メッセージタイプ）を指定している

  TextMessage	String
  MapMessage	キーがString、値がプリミティブ
  BytesMessage
  StreamMessage	プリミティブ値のストリーム
  ObjectMessage	Serializableオブジェクト
  Message	メッセージボディなし

JMS Queue Browsers

• キュー内のメッセージを調査するために使う
• キュー内のメッセージを見て、ヘッダーを表示する
• QueueBrowserの生成

  QueueBrowser browser = session.createBrowser(queue);
  

  • createBrowserの引数にメッセージセレクタを指定できる

JMS Exception Handling

• JMS APIの例外のハンドリングはJMSExceptionに対して行うといい

---

Creating Robust JMS Applications

• JMSの信頼性
  • メッセージの損失や重複を見過ごせない
  • メッセージはただ１度だけ受信される
  • トランザクション内で永続化されたメッセージ送信
  • トランザクション内でのメッセージ受信
  • メッセージに優先度の設定
  • メッセージ期限の設定

Using Basic Reliability Mechanisms

Controlling Message Acknowledgment

• メッセージ消費の成功の３ステップ
  1. クライアントがメッセージを受信
  2. クライアントがメッセージを処理
  3. メッセージの認識
     • 認識は、JMSプロバイダまたはクライアントによって初期化される
• トランザクションセッションの場合は、コミット時に自動的に認識がされる
  • ロールバックされた場合は、消費されたメッセージは再配送される
• 非トランザクションセッションの場合は、sessionFactory.createSessionの第2パラメータによっていつ、どのようにメッセージ認識されるかがことなる
  • Session.AUTO_ACKNOWLEDGE
    • クライアントがreceive()コールから正常受け取ったとき、MessageListenerが正常を返したとき、セッションはクライアントのメッセージ受信を認識する
    • 同期のreceive()コールの場合は、受信と認識が１ステップでおこなわれ、続いてメッセージ処理が行われる
  • Session.CLIENT_ACKNOWLEDGE
    • クライアントがmessage.acknowledge()を呼んで、メッセージを認識する
    • ある消費メッセージを認識すると、セッション内でそれまで消費したメッセージも認識される
  • Session.DUPS_OK_ACKNOWLEDGE
    • セッションがメッセージを遅延認識する
    • JMSプロバイダではメッセージ重複がおきえるため、重複メッセージを扱えるクライアントによって利用される
    • セッションが重複メッセージを避けるようにするためのオーバヘッドがへる
• セッションが終了したときにメッセージがキューから受信されて認識されなかったら、JMSプロバイダはメッセージを保持し、コンシューマがキューにアクセスしたときに再配送する
• プロバイダは、セッションクローズ時に未認識のメッセージを永続的なTopicSubscriberについても維持する
• プロバイダは、セッションクローズ時に未認識のメッセージを非永続的なTopicSubscriberについては破棄する
• キューまたは永続的なサブスクリプションを使っていれば、session.recoverで非トランザクションセッションを停止し、未認識メッセージとともに起動できる

Specifying Message Persistence

• JMSプロバイダが失敗したときに、メッセージが損失したかを判断する２つの配信モード
  • PERSISTENT
    • デフォルト
    • JMSプロバイダが失敗したときにメッセージがロストしないように扱われる
    • 送信されたメッセージはログに残される
  • NON_PERSISTENT
    • メッセージは保存されない
    • アプリケーション側で、メッセージロスト対策を考える
    • ストレージオーバヘッドがへり、パフォーマンスは向上する
• 配送モードの設定方法
  • producer.send(), producer.publish()の引数で指定

Setting Message Priority Levels

• JMSプロバイダが緊急のメッセージを優先して配送するようにできる
• 優先度は0-9でデフォルトは4
• 優先度の設定方法
  • producer.setPriority()
  • producer.send(), producer.publish()の引数で指定

Allowing Messages to Expire

• デフォルトではメッセージの有効期限切れなし
• メッセージ期限を設定する方法
  • producer.setTimeToLive()
  • producer.send(), producer.publish()の引数で指定

Creating Temporary Destinations

• たいていの長くつかわれるJMSデスティネーション（キュー、トピック）はプログラムからでなく管理的に作られる
• コネクションのスコープ内でのみ使われる一時的なデスティネーションを作れる
  • session.createTemporaryQueue(), session.createTemporaryTopic()
• 一時的デスティネーションを作成したコネクションから作られたコンシューマのみが、一時的デスティネーションからのメッセージを消費できる
• 任意のプロデューサは一時的デスティネーションにメッセージを送信できる
• 一時的デスティネーションが属するコネクションをクローズすると、一時的デスティネーションが削除され内容が消える
• 一時的デスティネーションを利用して、リクエスト／リプライを実現できる
  • プロデューサはメッセージ送信時に、JMSReplyToヘッダにデスティネーションを指定
  • コンシューマはメッセージのJMSReplyToヘッダに書かれているデスティネーションに返答する
  • コンシューマはリクエストJMSMessageIDヘッダの値をレスポンスJMSCorrelationIDに設定することで、元のリクエストを参照させることができる

Using Advanced Reliability Mechanisms

• Creating Durable Subscriptions
  • Pub/Subアプリケーションで、パブリッシュされたすべてのメッセージが受信されることを保証する
    • パブリッシャの配送モードをPERSISTENTにする
    • サブスクライバに対して永続サブスクリプションを使う
  • session.createConsumerで作成されるサブスクライバは、アクティブな時に配送されたメッセージしか受信できない
  • session.createDurableSubscriberで永続サブスクライバを作成できる
  • 永続サブスクライバはある時点で１つのアクティブなサブスクライバのみ
  • 永続サブスクライバJMSプロバイダによって保持された一意識別子を特定して、永続サブスクリプションを登録する
    • 同じ識別子をもつ後続のサブスクライバは、サブスクリプションを消費する
    • 永続サブスクライバがアクティブなサブスクライバを持たない場合は、JMSプロバイダがメッセージを保持する（サブスクライバに消費されるかメッセージ期限切れまで）
  • 永続サブスクライバの一意識別子
    • コネクションに対するクライアントID：クライアント固有のコネクションファクトリに管理的に付与
    • サブスクライバに対するトピックとサブスクリプションの名前：session.createDurableSubscriberの引数
  • コネクションをstartするとサブスクライバがアクティブになる
  • サブスクライバの停止はtopicSubscriber.close()で行う
  • サブスクライバがアクティブでなければ、JMSプロバイダはtopicに送信されたメッセージを保持する
    • 同じ（コネクションファクトリ、クライアントID、トピック、サブスクリプション名）で永続サブスクライバを作成すると、サブスクライバがアクティブになりメッセージが配送される
  • 永続サブスクライバを削除するにはsubscriber.close()とsession.unsubscribe()を行う
• Using JMS API LocalTransactions
  • メッセージ送受信に対してローカルトランザクションを利用できる
    • session.commit, session.rollback
    • commitはすべての生成されたメッセージが送信され、受信されたメッセージが認識されることを意味する
    • rollbackはすべての生成されたメッセージが破棄され、、受信されたメッセージが復帰し再送されることを意味する（期限切れでない場合）
  • トランザクション内で起動されたセッションはトランザクションセッションになる
  • commit,rollbackがよばれると、現在のトランザクションは終了し別のトランザクションが始まる
  • トランザクションセッションをcloseするとトランザクションはロールバックされる
  • トランザクション内でsendのみ、receiveのみ、receive→sendはOK
  • トランザクション内でrequest/replyはNG
    • トランザクションがコミットされてからsendされるため
  • 同じメッセージに対するsendとreceiveはトランザクションに含められない
    • トランザクションはクライアントとJMSプロバイダ間で発生するから
  • トランザクションは、同一セッションでのメッセージ送受信はOK。セッションをまたがり特定のメッセージの送受信はNG
  • createSessionのパラメータでトランザクションセッションかどうかを指定
  • 認識モードは非トランザクションセッションの場合に有効

    connection.createSession(true, 0);
    

  • セッションをメッセージリスナーに渡すことで、非同期での受信⇒送信を実現できる

---

Using the JMS API in Java EE Applications

• JMS APIをエンタープライズビーンアプリケーションやWebアプリケーションで使うには
• Web/EJBコンテナのコンポーネントは１コネクションあたり２個以上のアクティブなセッションを作らない
• アプリケーションクライアントでは１コネクションあたり複数のセッションがあっても良い

Using @Resource Annotations in Enterprise Bean or Web Components

• アプリケーションクライアントコンポーネントで@Resourceを使う場合は、普通はstaticでJMSリソースを参照する

  @Resource(lookup = "jms/ConnectionFactory")
  private static ConnectionFactory connectionFactory;
  
  @Resource(lookup = "jms/Queue")
  private static Queue queue;
  

• セッションビーン、メッセージ駆動ビーン、Webコンポーネントで@Resourceを使う場合はJMSリソースはstaticにしない

  @Resource(lookup = "jms/ConnectionFactory")
  private ConnectionFactory connectionFactory;
   
  @Resource(lookup = "jms/Topic")
  private Topic topic;
  

Using Session Beans to Produce and to Synchronously Receive Messages

• receive()は同期ブロッキングがかかるため、エンタープライズビーン内でreceive()を使うのはよくない
• メッセージを非同期で受け取るために、timed synchronous receiveかメッセージ駆動ビーンを用いる

Managing JMS Resources in Session Beans

• JMS APIリソース=JMS APIコネクション、セッション
• リソースは不要になったら解放する
  • ビジネスメソッド実行のスコープでリソースが必要なら、finallyブロックでリソースをclose
  • エンタープライズビーンのインスタンススコープでリソースを保持するなら、@PostConstructでリソース生成、@PreDestroyでclose
    • ステートフルセッションビーンでキャッシュの場合にもリソースを保持したい場合は、@PrePassivateでclose、@PostActivateでリソース生成

ManagingTransactions in Session Beans

• CMT（デフォルト）を使えば、EJBコンテナがトランザクション（メッセージの送受信含む）境界を管理する
• BMTで自前でUserTransactionを使ってトランザクション（メッセージの送受信含む）境界を管理もできるが、普通はCMT

Using Message-Driven Beans to Receive Messages Asynchronously

• セッションビーンを使うとメッセージの送信と、同期で受信処理できる
• メッセージ駆動ビーンを使うとJMSメッセージを非同期で受信処理できる
• メッセージ駆動ビーンはキューまたは永続的サブスクリプションからメッセージを消費するメッセージリスナ
  • メッセージはJavaEEコンポーネントまたは、JavaEE以外のアプリケーションから送信
• メッセージ駆動ビーンはメッセージが到着のコールバックonMessageをもつ
• メッセージ駆動ビーンのアプリケーションクライアントメッセージリスナとの違い
  • EJBコンテナによりセットアップ処理が実行される
    • メッセージを受信するコンシューマを作成
    • デプロイ時に、メッセージ駆動ビーンとコネクションファクトリ、デスティネーション、永続化サブスクリプション、メッセージセレクタが関連付けられる
    • メッセージリスナが登録される
    • 認識モードが設定される（デフォルトはAUTO_ACKNOWLEDGE）
  • @MessageDrivenでコネクションファクトリ、デスティネーションタイプ、永続化サブスクリプション、メッセージセレクタ、認識モードなどを指定
• メッセージ駆動ビーンはMessageListenerを実装する
• @PostConstructでコネクション生成、@PreDestroyでコネクションクローズ処理をいれてもいい
  • メッセージの送信
  • 別デスティネーションからの同期受信
• トランザクション管理のためにMessageDrivenContextをインジェクションできる
• メッセージ駆動ビーンはセッションビーンとは違って、ローカル／リモートインターフェースをもたない
• メッセージ駆動ビーンは生存期間が短く特定のクライアントむけの状態をもたない
• メッセージ駆動ビーンのインスタンス変数はクライアントメッセージを処理にわたっての状態をもつ
  • JMS APIコネクション
  • DBコネクション
  • エンタープライズビーンへの参照
• メッセージ駆動ビーンのインスタンスはコンテナによってプールされ、受信したメッセージを並行で処理できる
• コンテナは配送されたメッセージを先着順に処理しようとするが、かならずしも順序は保障されていない
  • アプリケーションはメッセージ到着順に依存しないように作成するべき
    • シーケンス性を保つためにメッセージをキャッシュ
    • 順序性をたもつために、送信者に確認メッセージを返す
• メッセージ駆動ビーンのインスタンスを作るためにコンテナは以下のタスクを実行
  • ビーンのインスタンスを作成
  • リソースインジェクション
  • @PostConstructメソッドの実行
• メッセージ駆動ビーンのインスタンスを削除するためにコンテは以下のタスクを実行
  • @PreDestroyメソッドの実行

Managing Distributed Transactions

• 分散トランザクションは、
  • 複数のアプリケーションが同一のDBに対してアトミックな更新をすることを可能にする
  • あるアプリケーションが複数のDBに対してアトミックな更新をすることを可能にする
• JMS APIを使うJavaEEアプリケーションでは、メッセージ送受信とDB更新やその他のリソース管理を統合できる
  • 単一トランザクションで複数のコンポーネントによるリソースにアクセスできる
• EJBコンテナの分散トランザクション管理は２種類
  • CMT
    • コンテナがトランザクションを管理するので、明示的なcommitやrollbackは不要
    • JMS APIを利用するときはおすすめ
    • エンタープライズビーンのトランザクション属性を指定する
    • あるJMSメッセージの送信⇒受信を同一トランザクションには入れられない
  • BMT
    • UserTransactionのcommitやrollbackを使って、明示的にトランザクション境界を指定
    • トランザクションのプログラミング経験者向け
  • メッセージ駆動ビーンでCMTもBMTも使える
    • トランザクション内ですべてのメッセージが受信され処理されることを保証したい場合は、CMTでRequiredトランザクション属性をつける（onMessageメソッド）
• メッセージ駆動ビーンでCMTを使っていると、MessageDrivenContextが利用できる
  • setRollbackOnly：エラーハンドリングで使用。例外が起きたらsetRollbackOnlyマークがつきトランザクションはロールバックされる
  • getRollbackOnly：現在のトランザクションにsetRollbackOnlyマークがついているかどうかを調べる
  • ふつうはビジネスメソッドのトランザクション属性Required
  • メッセージ認識はコンテナがコミットしたら自動的に行われるので、メッセージ認識モードを指定しない
• メッセージ駆動ビーンでBMTを使っていると、分散トランザクションコンテキストの外側で、onMessageメソッドへのメッセージ配送がおこなわれる
  • onMessageメソッドないでUserTransaction.begin～UserTransaction.commit|rollbackまでがトランザクション
  • 本来は、session.createSessionはトランザクション内で呼ばれないとNG
    • UserTransaction.rollbackを呼んでもメッセージが再配送されない
  • あるメッセージを処理するために複数のトランザクションを利用できる
  • メッセージ処理の一部だけトランザクションの外側で処理できる
• エンタープライズビーン内でセッションを作ると、コンテナによって引数は無視される

  connection.createSession(true, 0)
  

• onMessageメソッドでRuntime例外が発生すると、コンテナはメッセージを認識しないので、JMSプロバイダがメッセージを再送信する

Using the JMS API with Application Clients and Web Components

• JavaEEのアプリケーションクライアントは、メッセージの送信、同期受信、非同期受信（メッセージリスナ）ができる
• Webコンポーネントはメッセージの送信、同期受信ができるが、非同期受信はできない
  • ブロッキングが発生するので、Webコンポーネントで非同期受信はよくないのでTimed同期受信を利用する