BMTメッセージ駆動型BeanとSpringを使用したハイパフォーマンスなメッセージ処理
Pages: 1, 2, 3, 4, 5

再配信と再試行の方法

前述のアスペクトは、処理エラー（アプリケーションエラーや、JDBC、JMS、JCAなどのトランザクションリソースのエラーを含む）が発生した 場合に、適切にトランザクションをロールバックします。通常、メッセージングサーバ（WebLogic Server JMS、MQseriesなど）は、メッセージを再配信します。これにより、コンポーネントは再試行を実行します。何らかの理由で処理できない「不正な」 メッセージを受け取った場合は、無限ループに陥る可能性があります。

このようなケースを避けるため、メッセージ処理の再試行回数に上限を設けなければなりません。どのエンタープライズメッセージングサーバも、キュー ごとに再試行回数を設定できます。また、再試行回数が上限を超えた場合にとるべきアクション（たとえば、メッセージを特別な「デッドレター」キューに移す など）を定義することもできます。データベースサーバ、その他の必須リソースの停電によって、大量のメッセージが「デッドレター」に移される場合もありま す。このような場合、停電が解消すると、これらのメッセージは、再試行のために処理キューに戻されます。エンタープライズ監視アプリケーションの中には、 このような機能をサポートしているものもあります。

何らかの理由で、「デッドレター」キューを用意できない場合は、アプリケーション側に監視用の小さなアドバイスを実装します。これを、 MessageProcessorの実行フローの最初に、アドバイスとして実装することもできます。

実際の実装は、失敗したメッセージをどのように扱うかによって異なりますが、失敗している最中には利用できないリソースがあることを考慮しなければ なりません。たとえば、データベースは、このような時には利用できない場合があります。次の例は、失敗したメッセージを重大度付きでログファイルに出力し ます。このようなログを読み込んで、メッセージを再投稿するユーティリティを実装するのは、それほど難しくありません。

public final class MdbRetryCountAdvice 
                               implements MethodInterceptor {
  private Log log = LogFactory.getLog(getClass());
  private int retryCount;

  public void setMaxRetryCount(int retryCount) {
    this.retryCount = retryCount;
  }
  
  public Object invoke(MethodInvocation methodInvocation) 
                                               throws Throwable {
    Object o = methodInvocation.getArguments()[0];

    try {
      return methodInvocation.proceed();
    } catch (Throwable e) {
       
                          
MessageData data = (MessageData) o;
      if (data.getDeliveryCount() >= this.retryCount) {
        this.log.fatal("Retry count exceeded. Message discarded."
                                                      + data, e);
        return null;
      }
      // assume business logic logged exception already 
      // so no stack trace here.
      throw e;
    }
  }
}

これで、メッセージの重複を扱うメカニズムを持つことができたので、MDBをデプロイする際に、デフォルトの AUTO_ACKNOWLEDGEモードの代わりに、 DUPS_OK_ACKNOWLEDGEモー ドを使用することを検討します。これによって、MOMは配信時の重複を保証するためのリソースをほとんど消費しないので、理論的にはパフォーマンスが向上 するはずです。実際のパフォーマンスへの影響は、MOMプロバイダごとに異なりますので、実際の数値を確認するために負荷テストを実行することをお勧めし ます。IBM WebSphere MQの場合、これら2つのモード間でパフォーマンスに有意な差が認められませんでした。ただし、皆さんの環境では、結果が異なる可能性があります。

ビジネスオブジェクトプーリング

我々の方式は、メッセージの処理中にエラーが発生した場合に、 RuntimeExceptionが送出されることに依存 しています。EJB仕様によれば、この場合、J2EEサーバはMDBインスタンスを破棄しなければならないことを理解しておくことが重要です。我々の実装 では、具象MDBは極めて軽量（コードがほとんどない）なので、ほとんどの場合これは問題になりません。むしろ、MDBが使用している実際のビジネスオブ ジェクトに注意しなければなりません。様々な理由で（最も一般的なケースは、一般にスレッドセーフではないビジネスロジックプロセッサで、XMLパーサー を使用している場合）、シングルトン実装のビジネスオブジェクトを使用できない場合があります。このような場合、ビジネスオブジェクトを、Springコ ンテキストでシングルトンではなくプロトタイプとして定義しなければなりません。それは、このビジネスオブジェクトが複数のMDBインスタンスから同時に 呼び出されるかもしれないからです。

我々が設計した MessageDataDrivenBeanクラスは、シングルトンではないBeanを扱えるように既に最適化されています。ご承知のとおり、Beanの検索は、 onEjbCreate()メソッド内で、MDBインスタンスごとに1回実行されます。EJB仕様は、シングルスレッドのMDB実行を保証しているので、シングルトンでない検索を使用して、呼び出しの際のBeanインスタンス化コストを節約すべきです。 

<!-- Note that this defines a prototype -->  
<bean id="simpleProcessorImpl" singleton="false"
      class="org.javatx.mdb.SimpleMessageProcessor"/>

<bean id="simpleProcessor" 
      class="org.springframework.aop.framework.ProxyFactoryBean">
  <property name="targetSource" ref="simpleProcessorImpl"/>
  <property name="proxyInterfaces" 
            value="org.javatx.mdb.MessageProcessor"/>
  <property name="interceptorNames">
    <list>
      <idref local="mdbRetryCountAdvisor"/>
      <idref local="mdbTransactionAdvisor"/>
      <idref local="mdbDuplicateHandlingAdvisor"/>
      <idref local="messageProcessorPerformanceMonitorAdvisor"/>
      <idref local="messageProcessorTraceAdvisor"/>
    </list>
  </property>
</bean>

ビジネスオブジェクトBean（たとえば、XMLパーサー、その他、生成コストが高くつくオブジェクト）のインスタンスを過剰に生成するのを避け、 さらに最適化したい場合は、Springがサポートしているターゲットソースのプーリングを使用します。Springは、新たなメソッド呼び出しごとに、 同一インスタンスのプールを維持し、このプール内にオブジェクトを解放します。しかも、ソースコードを何も変更せずに、これを実現することができます。ア プリケーションコンテキスト内の SimpleProcessorを非シングルトンとして定義し、プールするターゲットソースを定義してから（プールサイズ、その他のプールパラメータもここで定義できます）、プールされたターゲットソースを使用するように simpleProcessorプロキシの定義を変更するだけです。 

<!-- Note that this defines a prototype -->  
<bean id="simpleProcessorProtImpl" singleton="false"
      class="org.javatx.mdb.SimpleMessageProcessor"/>

<bean id="simpleProcessorPooled" 
   class="org.springframework.aop.target.CommonsPoolTargetSource">
 <property name="targetBeanName" value="simpleProcessorProtImpl"/>
 <property name="maxSize" value="5"/>
</bean>>  

<bean id="simpleProcessor" 
      class="org.springframework.aop.framework.ProxyFactoryBean">
  <property name="targetSource" ref="simpleProcessorImpl"/>
  ...
</bean>

この簡単な変更によって、アプリケーションのパフォーマンスを向上させることができます。特に、メッセージ処理エラーが頻繁に発生することが予想される場合は効果があります。 onMessage()メソッドやそれに依存するメソッドから実行時の例外が送出されたときに、MDBインスタンスが破棄されるからです。Springは、プーリングの実装にJakarta Commons Poolを使用します。このため、アプリケーションに commons-pool jarを含めなければなりません。

ダウンロード

この論文で参照されているソースコードのダウンロードはこちらからできます： custom-mdb-processing-source.zip

まとめ

ここまで説明してきたとおり、我々は、once-and-only-onceのQoSを維持したまま、トランザクションなしのメッセージ消費を扱う という目標を無事に達成することができました。BMTに切り替えた結果、アプリケーションの振る舞いとトランザクションの扱いを今まで以上に制御できると 同時に、パフォーマンスを格段に向上させることができると確信しています。この変更の副作用として、メッセージを消費する際に、現在使用しているMOM が、XAトランザクションをしっかりサポートしているかどうかを心配する必要はなくなりました。

一見、コードの量と複雑さが増すように見えますが、この論文で示したとおり、このソリューションは、SpringのAOPフレームワークを使用して 構造化された方法で実装できます。これによって、独立したアドバイスを使用して、非機能要件の個々のアスペクトを構成できます。各アドバイスは、特定の機 能部分の処理を担当します。ただし、Springのアプリケーションコンテキストで一緒にアセンブルされるので、メッセージ処理のためのすべての非機能要 件（トランザクション処理、重複処理、不正メッセージ処理、および一般的な追跡、ログ、パフォーマンス監視などを含む）を管理できます。

関連情報

Dmitri Maximovichは、ソフトウェア設計、開発、および技術トレーニングを専門とする独立のコンサルタントです。彼は、この業界で12年以上の経験を持っており、当初からJ2EEに携わってきました。

Eugene Kuleshovは、独立のコンサルタントです。彼は、ソフトウェア設計および開発において12年以上の経験を持ち、アプリケーションセキュリティ、エンタープライズインテグレーション（EAI）、およびメッセージ指向ミドルウェアを専門としています。