6.ActiveMQについて

2007.02.14　株式会社四次元データ　宮澤了祐

6.1.埋め込みMessageBrokerの使用

Connectionオブジェクトを作成する際に自動的にサーバーを起動するように設定することが出来ます。

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
factory.setUseEmbeddedBroker(true);
Connection connection = factory.createConnection();

ActiveMQConnectionFactory#setUseEmbeddedBroker()メソッドを使用します。 このファクトリを使用してコネクションを作成した際にMessageBrokerが起動します。

一つのファクトリから複数のコネクションを作成しようとすると、コネクションごとにMessageBrokerを起動しようとするので注意が必要です。

また設定ファイルを使用したり、URLを設定する場合は次のようにします。

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();

factory.setBrokerXmlConfig("src/default.xml");
factory.setBrokerURL("vm://localhost");

factory.setUseEmbeddedBroker(true);
Connection connection = factory.createConnection();

setBrokerXmlConfig()メソッドで設定ファイルの設定を行い、setBrokerURL()メソッドでURLを設定します。 設定ファイルを認識するためにはこの両方のメソッドを実行する必要があります。

また設定ファイルを使用するためにはクラスパスに/lib/spring.jarを追加する必要があります。

6.2.ActiveMQConnectionFactoryの取得方法

JNDIの使用

java.naming.factory.initial = org.activemq.jndi.ActiveMQInitialContextFactory

# 標準コネクタのURL
java.naming.provider.url = vm://localhost

# 埋め込みMessageBrokerの使用
useEmbeddedBroker = true

# 設定ファイルへのパス
# これは埋め込みMessageBrokerで使用されます。
brokerXmlConfig = file:src/default.xml

# キューオブジェクトをJNDIに登録します。
# queue.[JNDI名] = [キューの名前]
queue.Queue = MyQueue

# トピックオブジェクトをJNDIに登録します。
# topic.[JNDI名] = [トピックの名前]
topic.Topic = MyTopic

jndi.propertiesファイルをクラスパスの通ったディレクトリに設置します。

JNDIを使用してDestinationオブジェクトを作成します。

InitialContext ic = new InitialContext();
ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory)ic.lookup("queueConnectionFactory");

QueueConnection connection = factory.createQueueConnection();
   
Queue myQueue = (Queue)ic.lookup("MyQueue");
Topic myTopic = (Topic)ic.lookup("MyTopic");		    			

6.3.メッセージの永続性

ActiveMQでは予期せぬエラーでMessageBrokerが終了してしまった場合などに、未送信のメッセージを保持する機能があります。

永続化の設定はbrokerのpersistence要素で設定します。 persistcence以下に様々な永続化の方法を設定します。

Persistenceを用いない設定

VMPersitenceを使用します。

<broker>
...
   <persistence>
      <vmPersistence />
   </persistence>
...
</broker>

MessageBrokerを終了した際に未送信のメッセージは消えてしまいます。

journalの使用

<broker>
...
  <persistence>
    <journalPersistence directory="../var/journal">
        <jdbcPersistence dataSourceRef="derby-ds"/>
    </journalPersistence>
  </persistence>
...
</broker>

journalを使用するには、データーベースをPersistenceとして設定する必要があります。

データーベースの使用

データーベースを使用するにはbeans要素以下にbean要素としてデーターベースの情報を設定する必要があります。

ActiveMQが対応しているデーターベースには次があります。

• Apache Derby
• Oracle
• Sybase
• DB2
• SQLServer
• Postgresql
• MySQL
• Axion
• HSQL

以下はActiveMQ内部にあるApacheDerbyデーターベースを使用した例です。

<beans>
    <broker>
        <persistence>
            <jdbcPersistence dataSourceRef="derby-ds"/>
        </persistence>
    </broker>

    <bean id="derby-ds"
        class="org.apache.commons.dbcp.BasicDataSource"
        destroy-method="close">
    <property name="driverClassName">
      <value>org.apache.derby.jdbc.EmbeddedDriver</value>
    </property>
    <property name="url">
      <value>jdbc:derby:derbydb</value>
    </property>
    <property name="username">
      <value></value>
    </property>
    <property name="password">
      <value></value>
    </property>
    <property name="poolPreparedStatements">
      <value>true</value>
    </property>
  </bean>
</beans>

jdbcPersistece要素のdataSourceRef属性に設定したbeanのidを入力します。

6.4.ActiveMQ4.0系の変更点と新機能

2005年11月9日にActiveMQ4.0がリリースされました。 このバージョンでの変更点と新機能について主なものを 紹介します。

変更点

以前のバージョンからはいくつかの変更点があり、利用するには注意が必要です。

• パッケージ名の変更
  Apacheプロジェクトの一部に組み込まれたため、パッケージ名が「org.activemq」から「org.apache.activemq」に変更になりました。
• 設定ファイルの変更
  XML設定ファイルのスキーマが、XSDに変更になりました。
• 冗長化の方法の変更
  MessageBrokerを冗長化する際に使用するURIが「reliable」から「failover」に変更になりました。
• 埋め込みMessageBrokerの変更
  ActiveMQ4.xで埋め込みMessageBrokerを使用するには次のようにします。

  ActiveMQConnectionFactory factory = 
         new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
  

• XBeanの導入
  Springを使用していたBrokerFactoryですが、Springの使用をやめ、XBeanを使うようになりました。
• MessageBroker間の通信の方式の変更
  Broker間の通信をJMSのAPIを使用せず、低レベルなAPIを用いて通信を行うようになり、JMS間の通信を行うコードを記述する必要がなくなりました。。

新機能

ActiveMQ4.xで新たに追加された機能について解説します。

• Master Slave MessageBroker
  「reliable://」プロトコルが「failover://」プロトコルに変更になり、MasterSlave方式のレプリケートが可能になりました。
• Exclusive Consumer
  Exclusive ConsumerはひとつのMessageConsumer（Queueなど）のみにメッセージが送られることを保障するものです。
• Message Group
  Messageに対してJMSXGroupIDプロパティを割り振ることで、グループ化を行うことが出来ます。
  例えば次のように使用します。

  Mesasge message = session.createTextMessage("hoge");
  message.setStringProperty("JMSXGroupID", "TECHSCORE_HOGE");
  ...
  producer.send(message);
  

  グループ化されたMessageはそのグループ内での順番が保障されます。
• JAAS認証機能の追加
• Consumer Priority
  MessageConsumerに優先度をつけることが出来るようになりました。
• Retroactive Consumer
  MessageConsumerがQueueやTopicに対して購読を申し込んだ際に、購読を申し込む以前のメッセージを受け取ることが可能になりました。