JBossMQを理解するための簡単なサンプル。

JBossMQを理解するための簡単なサンプル。

 ※JBossのバージョンは、jboss4.0.3SP1 を利用しました。

◆JBossMQとは？
  JBossに組み込まれたプロセス間での情報伝達を確実にしてくれる仕組み。
  発信者と受信者の間に入って情報の媒介を行います。
  これにより、プロセス間のデータの未到達を防ぐことができます。

◆TOPICとQUEUE
  TOPIC、QUEUEともに情報伝達の媒介者のことである。
  TOPICは１対多への通信を媒介し、QUEUEは１対１の通信を媒介する。
  TOPICは発信者から情報を受け取ると即座に多数の受信者へ情報を配信し、
  QUEUEは発信者から情報を受け取ると即座に特定の受信者に情報を発信する。
  また、QUEUEは発信者が待ち受け状態にない場合は、次回の待ち受け状態まで情報を保持する。

◆JBossMQの起動
  JBOSS_HOME/bin/run.bat をキックすれば、jboss と同時に jbossmq も起動する。
  何の設定もしなくても、デフォルトでTOPICとQUEUEが用意されているので楽です。
  あらかじめ用意されているTOPICとQUEUEは以下のとおり。

  TOPIC名: topic/testTopic
  QUEUE名: queue/A

◆発信者(Publisher, Sender)
  TOPICに対して発信する者を Publisher という。
  QUEUEに対して発信する者を Sender という。

◆受信者(Subscriber, Receiver)
  TOPICから受信する者を Subscriber という。
  QUEUEから受信する者を Receiver という。

◆通信可狽ﾈデータ型

  Text のみならず全ての Object をやり取りできます。
  ただしObjectをやり取りする場合は Serializable にする必要があります。




--------------------------------------------------------------------------------------
◆TOPICを使うサンプルメ[ス[Publisher]
--------------------------------------------------------------------------------------
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import java.io.Serializable;

import javax.jms.TopicConnectionFactory;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.TopicPublisher;
import javax.jms.Topic;
import javax.jms.TextMessage;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.JMSException;
/**
 * testTopicトピックへテキストメッセージをパブリッシュする簡単なJMSクライアント
 */
public class HelloPublisher {

  /**
   * トピック接続です。クローズできるようにこれを取っておきます。
   */
  TopicConnection topicConnection;

  /**
   * トピックセッションです。クローズできるようにこれを取っておきます。
   * Also used to create messages.
   */
  TopicSession topicSession;

  /**
   * メッセージをパブリッシュするためにこれを使います。
   */
  TopicPublisher topicPublisher;

  /**
   * パブリッシュする宛先
   */
  Topic topic;


  /**
   * JMSにおけるお決まりの設定をします。
   *
   * オブジェクトの終了時にclose()を使います。
   *
   * @param factoryJNDI ルックアップするトピックの接続ファクトリ名
   * @param topicJNDI ルックアップするトピックの宛先名
   */
  public HelloPublisher(String factoryJNDI, String topicJNDI)
    throws JMSException, NamingException {

    // イニシャルコンテキストを取得します。
    Context context = new InitialContext();

    // 接続ファクトリを取得します。
    TopicConnectionFactory topicFactory =
      (TopicConnectionFactory)context.lookup(factoryJNDI);

    // 接続を作成します。
    topicConnection = topicFactory.createTopicConnection();

    // セッションを作成します。
    topicSession = topicConnection.createTopicSession(
      // トランザクションなし
      false,
      // 自動応答確認
      Session.AUTO_ACKNOWLEDGE);

    // 宛先をルックアップします。
    topic = (Topic)context.lookup(topicJNDI);

    // パブリッシャを作成します。
    topicPublisher = topicSession.createPublisher(topic);

  }

  /**
   * testTopicトピックへのJMSメッセージとして与えられた文字列をパブリッシュします。
   */
  public void publish(Serializable msg) throws JMSException {

    // メッセージを作成します。
    ObjectMessage message = topicSession.createObjectMessage();
    message.setObject(msg);

    // メッセージをパブリッシュします。
    topicPublisher.publish(topic, message);

  }

  /**
   * セッションと接続をクローズします。
   * 終了したら、パブリッシュはもうできません。
   */
  public void close() throws JMSException {

    topicSession.close();
    topicConnection.close();

  }

  /**
   * testTopicへ10個のメッセージをパブリッシュする例を実行します。
   * JBoss 2.4.0以上でのみ動作します。
   */
  public static void main(String[] args) {
    try {

      // HelloPublisherを作成します。引数として、TopicConnection
      // ファクトリの名前とルックアップのためのTopicの宛先を渡します。
      HelloPublisher publisher = new HelloPublisher(
        // ConnectionFactoryの名前
        "TopicConnectionFactory",
        // パブリッシュする宛先の名前
        "topic/testTopic");

      // 10個のメッセージをパブリッシュします。
      for (int i = 1; i < 11; i++) {

        String msg = "Hello World no. " + i;
        System.out.println("Publishing message: " + msg);
        publisher.publish((Serializable)msg);

      }

      // パブリッシャをクローズします。
      publisher.close();

    } catch(Exception ex) {

      System.err.println(
        "An exception occurred while testing HelloPublisher: " + ex);
      ex.printStackTrace();

    }

  }

} // HelloPublisher


--------------------------------------------------------------------------------------
◆TOPICを使うサンプルメ[ス[Subscriber]
--------------------------------------------------------------------------------------
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import java.io.Serializable;

import javax.jms.TopicConnectionFactory;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.jms.Topic;
import javax.jms.Message;
import javax.jms.TextMessage;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.MessageListener;
import javax.jms.JMSException;

/**
 * testTopicトピックへのメッセージのためにサブスクライブする簡単なJMSクライアント
 */
public class HelloSubscriber implements MessageListener {

  /**
   * トピック接続です。 クローズできるようにこれを取っておきます。
   */
  TopicConnection topicConnection;

  /**
   * トピックセッションです。クローズできるようにこれを取っておきます。
   * メッセージを作成するのにも使われます。
   */
  TopicSession topicSession;

  /**
   * サブスクライバ
   */
  TopicSubscriber topicSubscriber;

  /**
   * サブスクライブする宛先
   */
  Topic topic;

  /**
   * JMSにおけるお決まりの設定をします。
   *
   * オブジェクトの終了時にclose()を使います。
   *
   * @param factoryJNDI ルックアップするトピックの接続ファクトリ名
   * @param topicJNDI ルックアップするトピックの宛先名
   */
  public HelloSubscriber(String factoryJNDI, String topicJNDI)
    throws JMSException, NamingException
  {

    // イニシャルコンテキストを取得します。
    Context context = new InitialContext();

    // 接続ファクトリを取得します。
    TopicConnectionFactory topicFactory =
      (TopicConnectionFactory)context.lookup(factoryJNDI);

    // 接続を作成します。
    topicConnection = topicFactory.createTopicConnection();

    // セッションを作成します
    topicSession = topicConnection.createTopicSession(
      // トランザクションなし
      false,
      // 自動応答確認
      Session.AUTO_ACKNOWLEDGE);

    // 宛先をルックアップします。
    topic = (Topic)context.lookup(topicJNDI);

    // サブスクライバを作成します。
    topicSubscriber = topicSession.createSubscriber(topic);

    // メッセージリスナを設定します。それはMessageListenerインタフェースを
    // 実装したクラスです。
    topicSubscriber.setMessageListener(this);

    System.out.println(
      "HelloSubscriber subscribed to topic: " + topicJNDI);

    // 任意のメッセージを受信するメッセージリスナのために、その接続は開始
    // させられなければなりません。
    topicConnection.start();
  }

  /**
   * MessageListerインタフェースの実装
   * メッセージはこのメャbドを使って受信されます。
   */
  public void onMessage(Message m) {

    // 正しいメッセージタイプにキャストするときに注意して、
    // メッセージの中身を出します。onMessageはアプリケーション例外を
    // スローすべきではありません。
    try {

      Object msg = ((ObjectMessage)m).getObject();
      System.out.println("HelloSubscriber got message: " + msg.toString());

    } catch(JMSException ex) {

      System.err.println("Could not get text message: " + ex);
      ex.printStackTrace();

    }

  }

  /**
   * セッションと接続をクローズします。
   */
  public void close() throws JMSException {

    topicSession.close();
    topicConnection.close();

  }

  /**
   * トピックtestTopicへサブスクライブする例を実行します。
   * JBoss 2.4.0以上でのみ動作します。
   */
  public static void main(String[] args) {

    try {

      // HelloSubscriberを作成します。引数として、TopicConnectionの
      // ファクトリの名前とルックアップのためのTopicの宛先を指定します。
      HelloSubscriber subscriber = new HelloSubscriber(
        // ConnectionFactoryの名前
        "TopicConnectionFactory",
        // パブリッシュする宛先名
        "topic/testTopic");

    } catch(Exception ex) {

      System.err.println(
        "An exception occurred while testing HelloSubscriber: " + ex);
      ex.printStackTrace();

    }

  }

} // HelloSubscriber



--------------------------------------------------------------------------------------
◆QUEUEを使うサンプルメ[ス[Sender]
--------------------------------------------------------------------------------------
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import java.io.Serializable;

import javax.jms.QueueConnectionFactory;
import javax.jms.QueueConnection;
import javax.jms.QueueSession;
import javax.jms.QueueSender;
import javax.jms.Queue;
import javax.jms.TextMessage;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.JMSException;
/**
 * testQueueトピックへテキストメッセージをパブリッシュする簡単なJMSクライアント
 */
public class HelloSender {

  /**
   * トピック接続です。クローズできるようにこれを取っておきます。
   */
  QueueConnection queueConnection;

  /**
   * トピックセッションです。クローズできるようにこれを取っておきます。
   * Also used to create messages.
   */
  QueueSession queueSession;

  /**
   * メッセージをパブリッシュするためにこれを使います。
   */
  QueueSender queueSender;

  /**
   * パブリッシュする宛先
   */
  Queue queue;


  /**
   * JMSにおけるお決まりの設定をします。
   *
   * オブジェクトの終了時にclose()を使います。
   *
   * @param factoryJNDI ルックアップするトピックの接続ファクトリ名
   * @param queueJNDI ルックアップするトピックの宛先名
   */
  public HelloSender(String factoryJNDI, String queueJNDI)
    throws JMSException, NamingException {

    // イニシャルコンテキストを取得します。
    Context context = new InitialContext();

    // 接続ファクトリを取得します。
    QueueConnectionFactory queueFactory =
      (QueueConnectionFactory)context.lookup(factoryJNDI);

    // 接続を作成します。
    queueConnection = queueFactory.createQueueConnection();

    // セッションを作成します。
    queueSession = queueConnection.createQueueSession(
      // トランザクションなし
      false,
      // 自動応答確認
      Session.AUTO_ACKNOWLEDGE);

    // 宛先をルックアップします。
    queue = (Queue)context.lookup(queueJNDI);

    // パブリッシャを作成します。
    queueSender = queueSession.createSender(queue);

  }

  /**
   * testQueueトピックへのJMSメッセージとして与えられた文字列をパブリッシュします。
   */
  public void send(Serializable msg) throws JMSException {

    // メッセージを作成します。
    ObjectMessage message = queueSession.createObjectMessage();
    message.setObject(msg);

    // メッセージをパブリッシュします。
    queueSender.send(queue, message);

  }

  /**
   * セッションと接続をクローズします。
   * 終了したら、パブリッシュはもうできません。
   */
  public void close() throws JMSException {

    queueSession.close();
    queueConnection.close();

  }

  /**
   * testQueueへ10個のメッセージをパブリッシュする例を実行します。
   * JBoss 2.4.0以上でのみ動作します。
   */
  public static void main(String[] args) {
    try {

      // HelloSenderを作成します。引数として、QueueConnection
      // ファクトリの名前とルックアップのためのQueueの宛先を渡します。
      HelloSender sender = new HelloSender(
        // ConnectionFactoryの名前
        "QueueConnectionFactory",
        // パブリッシュする宛先の名前
        "queue/A");

      // 10個のメッセージをパブリッシュします。
      for (int i = 1; i < 11; i++) {

        String msg = "Hello World no. " + i;
        System.out.println("Sending message: " + msg);
        sender.send((Serializable)msg);

      }

      // パブリッシャをクローズします。
      sender.close();

    } catch(Exception ex) {

      System.err.println(
        "An exception occurred while testing HelloSender: " + ex);
      ex.printStackTrace();

    }

  }

} // HelloSender

--------------------------------------------------------------------------------------
◆QUEUEを使うサンプルメ[ス[Receiver]
--------------------------------------------------------------------------------------

import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import java.io.Serializable;

import javax.jms.QueueConnectionFactory;
import javax.jms.QueueConnection;
import javax.jms.QueueSession;
import javax.jms.QueueReceiver;
import javax.jms.Queue;
import javax.jms.Message;
import javax.jms.TextMessage;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.MessageListener;
import javax.jms.JMSException;

/**
 * testQueueトピックへのメッセージのためにサブスクライブする簡単なJMSクライアント
 */
public class HelloReceiver implements MessageListener {

  /**
   * トピック接続です。 クローズできるようにこれを取っておきます。
   */
  QueueConnection queueConnection;

  /**
   * トピックセッションです。クローズできるようにこれを取っておきます。
   * メッセージを作成するのにも使われます。
   */
  QueueSession queueSession;

  /**
   * サブスクライバ
   */
  QueueReceiver queueReceiver;

  /**
   * サブスクライブする宛先
   */
  Queue queue;

  /**
   * JMSにおけるお決まりの設定をします。
   *
   * オブジェクトの終了時にclose()を使います。
   *
   * @param factoryJNDI ルックアップするトピックの接続ファクトリ名
   * @param queueJNDI ルックアップするトピックの宛先名
   */
  public HelloReceiver(String factoryJNDI, String queueJNDI)
    throws JMSException, NamingException
  {

    // イニシャルコンテキストを取得します。
    Context context = new InitialContext();

    // 接続ファクトリを取得します。
    QueueConnectionFactory queueFactory =
      (QueueConnectionFactory)context.lookup(factoryJNDI);

    // 接続を作成します。
    queueConnection = queueFactory.createQueueConnection();

    // セッションを作成します
    queueSession = queueConnection.createQueueSession(
      // トランザクションなし
      false,
      // 自動応答確認
      Session.AUTO_ACKNOWLEDGE);

    // 宛先をルックアップします。
    queue = (Queue)context.lookup(queueJNDI);

    // サブスクライバを作成します。
    queueReceiver = queueSession.createReceiver(queue);

    // メッセージリスナを設定します。それはMessageListenerインタフェースを
    // 実装したクラスです。
    queueReceiver.setMessageListener(this);

    System.out.println(
      "HelloReceiver subscribed to queue: " + queueJNDI);

    // 任意のメッセージを受信するメッセージリスナのために、その接続は開始
    // させられなければなりません。
    queueConnection.start();
  }

  /**
   * MessageListerインタフェースの実装
   * メッセージはこのメャbドを使って受信されます。
   */
  public void onMessage(Message m) {

    // 正しいメッセージタイプにキャストするときに注意して、
    // メッセージの中身を出します。onMessageはアプリケーション例外を
    // スローすべきではありません。
    try {

      Object msg = ((ObjectMessage)m).getObject();
      System.out.println("HelloReceiver got message: " + msg.toString());

    } catch(JMSException ex) {

      System.err.println("Could not get text message: " + ex);
      ex.printStackTrace();

    }

  }

  /**
   * セッションと接続をクローズします。
   */
  public void close() throws JMSException {

    queueSession.close();
    queueConnection.close();

  }

  /**
   * トピックtestQueueへサブスクライブする例を実行します。
   * JBoss 2.4.0以上でのみ動作します。
   */
  public static void main(String[] args) {

    try {

      // HelloReceiverを作成します。引数として、QueueConnectionの
      // ファクトリの名前とルックアップのためのQueueの宛先を指定します。
      HelloReceiver receiver = new HelloReceiver(
        // ConnectionFactoryの名前
        "QueueConnectionFactory",
        // パブリッシュする宛先名
        "queue/A");

    } catch(Exception ex) {

      System.err.println(
        "An exception occurred while testing HelloReceiver: " + ex);
      ex.printStackTrace();

    }

  }

} // HelloReceiver
--------------------------------------------------------------------------------------

◆サンプルのコンパイル。(同じ階層に jbossall-client.jar があることが前提です。)

  javac -classpath ;.;./jbossall-client.jar; *.java

◆サンプルの起動方法。(同じ階層に jbossall-client.jar があり、jbossmq がデフォルトで起動していることが前提です。)

  java -classpath ;.;./jbossall-client.jar; HelloPublisher
  java -classpath ;.;./jbossall-client.jar; HelloSubscriber
  java -classpath ;.;./jbossall-client.jar; HelloSender
  java -classpath ;.;./jbossall-client.jar; HelloReceiver

◆サンプルの起動結果。

  Subscriber もしくは Receiver のコンメ[ルに以下のように侮ｦされます。

   HelloSubscriber got message: Hello World no. 1
   HelloSubscriber got message: Hello World no. 2
   HelloSubscriber got message: Hello World no. 3
   HelloSubscriber got message: Hello World no. 4
   HelloSubscriber got message: Hello World no. 5
   HelloSubscriber got message: Hello World no. 6
   HelloSubscriber got message: Hello World no. 7
   HelloSubscriber got message: Hello World no. 8
   HelloSubscriber got message: Hello World no. 9
   HelloSubscriber got message: Hello World no. 10


◆参考サイト。

  JMSプロバイダー
  http://neverbird.sourceforge.jp/manual/fancy/ch08s07.html