受信処理のためのメッセージのバッチ処理

バッチコールバック

メッセージング エンジンに受信アダプターによって送信されたバッチは非同期的に処理されます。 そのため、アダプターには、成功または失敗の通知を受け取り、必要なクリーンアップ アクションを実行できるように、コールバックをアダプター内の何らかの状態に関連付けるメカニズムが必要です。 コールバック セマンティクスは、アダプターが 1 つまたは 3 つのアプローチの組み合わせを使用できるように柔軟です。 これらは：

• すべてのコールバックは、 IBTBatchCallBack を実装する同じオブジェクト インスタンスで行われます。

• 新しいバッチを要求するときにエンジンに渡される Cookie は、コールバックをアダプター内の状態に関連付けるために使用されます。

• IBTBatchCallBack を実装するバッチごとに異なるコールバック オブジェクトがあります。 ここでは、各オブジェクトは独自のプライベート状態を保持します。

  バッチが処理されると、アダプターは IBTBatchCallBack.BatchComplete の実装で呼び出されます。 バッチの全体的な状況は、最初のパラメーター HRESULT 状況によって示されます。 この値が 0 以上の場合、バッチは、エンジンがデータの所有権を持ち、アダプターがそのデータをワイヤから自由に削除できるという意味で成功しました。 負の状態は、バッチが失敗したことを示します。バッチ内のどの操作も成功しなかったため、アダプターはエラーを処理します。

  バッチが失敗した場合、アダプターは操作が失敗した項目を認識する必要があります。 たとえば、アダプターがディスクから 20 個のファイルを読み取り、1 つのバッチを使用して BizTalk Server に送信したとします。 10 番目のファイルが破損している場合、アダプターはその 1 つのファイルを中断し、残りの 19 を再送信する必要があります。 この情報は、アダプターが 2 番目と 3 番目のパラメーターで使用できます。opCount (操作数) は short 型と operationStatus 型で、BTBatchOperationStatus[] 型です。

操作数は、バッチ内の操作の種類の数 ( BTBatchOperationStatus 配列のサイズ) を示します。 操作状態配列内の各要素は、指定された操作の種類に対応します。 アダプターは、BTBatchOperationStatus 配列を使用して、エラーを示す負の HRESULT 値の BTBatchOperationStatus.MessageStatus 配列を調べることで、特定の操作で失敗した項目を特定できます。 上記のシナリオでは、アダプターは、19 件のメッセージ送信と 1 つのメッセージ中断を含む新しいバッチを作成します。

次のコード フラグメントは、アダプターがトランスポート プロキシを介してエンジンから新しいバッチを要求し、Cookie アプローチを使用して 1 つのメッセージをエンジンに送信する方法を示しています。 メッセージング エンジンは、送信されたメッセージのバッチ全体の処理が完了すると、バッチ コールバックとして BatchComplete メソッドを呼び出します。

using Microsoft.BizTalk.TransportProxy.Interop;  
using Microsoft.BizTalk.Message.Interop;  
  
public class MyAdapter :   
                  IBTTransport,   
                  IBTTransportConfig,   
                  IBTTransportControl,  
                  IPersistPropertyBag,   
                  IBaseComponent,  
                  IBTBatchCallBack  
{  
      private IBTTransportProxy _tp;  
  
      public void BatchComplete(      
            Int32                         status,   
            Int16                         opCount,   
            BTBatchOperationStatus[]      operationStatus,   
            System.Object                 callbackCookie)  
      {  
            // Use cookie to correlate callback with work done,  
            // in this example the batch is to submit a single  
            // file the name of which will be in the  
            // callbackCookie  
            string fileName = (string)callbackCookie;  
            if ( status >= 0 )  
                  // DeleteFile from disc  
                  File.Delete(fileName);          
            else  
                  // Rename file to fileName.bad  
                  File.Move(fileName, fileName + ".bad");  
      }  
  
      private void SubmitMessage(  
            IBaseMessage                  msg,   
            string                        fileName)  
      {  
            // Note: Pass in the filename as the cookie  
            IBTTransportBatch batch =   
                  _tp.GetBatch(this, (object)fileName);  
  
            // Add msg to batch for submitting   
            batch.SubmitMessage(msg);   
  
            // Process this batch  
            batch.Done(null);  
      }  
}  

BizTalk Server SDK には、ファイル、HTTP、MSMQ、トランザクション アダプターの各アダプターのサンプルが含まれています。 これらのアダプターはすべて、BaseAdapter と呼ばれる共通の構成要素の上に構築されます。 BaseAdapter のバージョン 1.0.1 には、操作の状態を解析し、送信する新しいバッチを再構築するための関連するすべてのコードが含まれています。

競合状態

エラーを解決し、送信されたバッチの最終的な結果を決定する 2 つのタスクは簡単に見えますが、実際には異なるスレッドからの情報に依存しています。

• アダプターは、BizTalk Server からアダプターの BatchComplete コールバック メソッドに渡された情報に基づいてエラーを処理します。 このコールバックは、アダプターのスレッドで実行されます。

• DTCCommitConfirm は、 IBTDTCCommitConfirm オブジェクトのメソッドです。 IBTDTCCommitConfirm オブジェクトのインスタンスは、バッチ IBTTransportBatch::D one 呼び出しによって返されます。 このインスタンスは、IBTTransportBatch::Done 呼び出しと同じスレッド上にあり、これはアダプターのコールバック スレッドとは異なります。

  アダプターが IBTTransportBatch::D に対して行うすべての呼び出しについて、メッセージング エンジンは、バッチ送信の結果を報告するために、別のスレッドでコールバック メソッド BatchComplete に対応する呼び出しを行います。 BatchComplete では、アダプターは、バッチが成功したか失敗したかに基づいて、トランザクションをコミットまたはロールバックする必要があります。 どちらの場合も、アダプターは DTCCommitConfirm を呼び出してトランザクションの状態を報告する必要があります。

  アダプターの BatchComplete の実装では、IBTTransportBatch::D one によって返される IBTDTCCommitConfirm オブジェクトが BatchComplete の実行時に常に使用可能であると想定できるため、競合状態が発生する可能性があります。 ただし、IBTTransportBatch::Done が戻る前であっても、BatchComplete は別のメッセージング エンジン スレッドで呼び出すことができます。 アダプターが BatchComplete 実装の一部として IBTDTCCommitConfirm オブジェクトにアクセスしようとすると、呼び出し元のスレッドが存在しなくなったため、アクセス違反が発生する可能性があります。 この状態を回避するには、次の解決策を使用します。

  次の例では、イベントを使用して問題を解決しています。 ここでは、イベントを使用するプロパティを介してインターフェイス ポインターにアクセスします。 get は、先に進む前にセットが完了するまで常に待機します。

protected IBTDTCCommitConfirm CommitConfirm  
{  
      set  
      {  
            this.commitConfirm = value;  
            this.commitConfirmEvent.Set();  
      }  
      get  
      {  
            this.commitConfirmEvent.WaitOne();  
            return this.commitConfirm;  
      }  
}  
protected IBTDTCCommitConfirm commitConfirm = null;  
private ManualResetEvent commitConfirmEvent = new ManualResetEvent(false);  

バッチ状態コード

バッチ全体の状態コードは、バッチの結果を示します。 操作の状態は、個々のメッセージ アイテムの送信状態コードを示します。 バッチは、さまざまな理由で失敗する可能性があります。 セキュリティ関連のエラーが発生したか、エンジンのシャットダウン時にメッセージが送信された可能性があります。 (通常、エンジンがシャットダウンすると、新しい作業は受け入れられていませんが、インプロセス作業の完了は許可されます)。次の表は、バッチ状態または操作状態のいずれかで返される一般的な HRESULT 値を示しています。 また、成功コードと失敗コードのどちらであるかを示します。 これらのコードの適切な処理については、「 アダプターの障害を処理する方法」で詳しく説明されています。

バッチ操作

次の表では、アダプターが特定のバッチに作業を追加するために使用する IBTTransportBatch オブジェクトのメンバー メソッドと操作について詳しく説明します。

バッチ管理

バッチは、メッセージング エンジンのネイティブ コードで実装されます。 このため、マネージド コードで記述されたアダプターは、バッチのランタイム呼び出し可能ラッパー (RCW) を、バッチの処理が完了した後に解放する必要があります。 これは、 Marshal.ReleaseComObject API を使用してマネージド コードで行われます。 返された参照カウントが 0 に達するまで、この API を while ループで呼び出す必要があることを覚えておくことをお勧めします。

BizTalk Server は、バッチ内でメッセージを同期的に処理します。 多くのバッチが同時に処理される場合があり、アプリケーション ドメインのバッチ サイズを調整することで、アダプター内の何らかの最適化の機会を提供する可能性があります。 たとえば、FTP サイト上のすべてのファイルを (何らかの制限に達するまで) 処理できます。 SAP の場合は、1 つのストリームを多数のメッセージに処理し、バッチとして送信することができます。

BizTalk Server に操作を渡すためにアダプターによって使用されるバッチは、BizTalk Server がアダプターに提供するメッセージの一覧に正確に対応する必要はありません。 つまり、トランザクション送信を実行する場合は、再送信操作 MoveToNextTransport と MoveToSuspendQ を別々のバッチに分割する必要があります。 多くのアダプターは、複数のエンドポイントに送信されたメッセージのバッチを、さらに処理するために個別のメッセージのリストに並べ替えます。

ポイントは、BizTalk Server でのメッセージのバッチ処理に関連付けられているルール (トランザクションに関連付けられているルール以外) が存在しないということです。 バッチ処理は、アダプターが BizTalk Server との間でメッセージをチャンクする実装固有の方法です。

アダプターは、BizTalk Server が BizTalk Server への応答のためにより大きなバッチに渡した複数の小さなバッチからのメッセージをバッチ処理できます。 これは、多数の非常に小さなメッセージを処理するトランザクション アダプターの大幅な最適化である可能性があります。 "メッセージあたりのトランザクションの合計数" 比率が最小限に抑えられます。

通常、BizTalk Server は、5 ~ 10 個のメッセージの送信側バッチを生成します。 これらのメッセージが非常に小さい場合、アダプターは、BizTalk Server に削除のトランザクション バッチを送信する前に、最大 100 メッセージ以上をバッチ処理する可能性があります。 このような最適化は実装が容易ではありません。メッセージがアダプター メモリにスタックしないようにし、しきい値が満たされるまで無限に待機する必要があります。

バッチ処理の重要性は、MQSeries などの高スループット アダプターの BizTalk Server のパフォーマンス数値で確認できます。 これらのアダプターでは、メッセージは送信の少なくとも 2 倍の頻度で受信されます。 既定では、受信アダプターは 100 個のメッセージのバッチを使用し、送信アダプターは指定された既定の BizTalk Server バッチを使用します。

トランザクション バッチ

トランザクション オブジェクトを作成し、それを BizTalk Server に提供するアダプターを作成する場合、次の処理を行うコードを記述する責任を負います。

• バッチ操作の最終的な結果を決定します。トランザクションをコミットまたは終了します。 これは、メッセージ キュー (MSMQ) キューへの書き込みやトランザクション データベース操作など、BizTalk Server に依存しない、この 1 つのトランザクションのスコープ内の他のトランザクション ブランチによって異なる場合があります。

• IBTDTCCommitConfirm.DTCCommitConfirm メソッドを呼び出して、最終的な結果を BizTalk Server に通知します。 trueの戻り値は、トランザクションの正常なコミットを示します。falseを返す場合は、トランザクションの失敗とロールバックが示されます。

  アダプターは、内部追跡データを維持するために、トランザクションの最終的な結果について BizTalk Server に通知する必要があります。 アダプターは、 DTCCommitConfirm を呼び出すことによって結果を BizTalk Server に通知します。 アダプターがこれを行わない場合、大量のメモリ リークが発生し、MSDTC トランザクションがタイムアウトし、操作が正常に完了したにもかかわらず失敗する可能性があります。