第10章 非同期継続

jBPM は グラフ指向プログラミング (GOP) をベースとしています。基本的に、GOP では、同時実行パスを処理できる単純状態マシンを指定しますが、特定の実行アルゴリズムでは、すべての状態遷移が単一スレッドのクライアント操作で行われます。デフォルトでは、クライアントのスレッドで状態遷移を実行するのが適切なアプローチです。このアプローチは、サーバー側のトランザクションに自然に適合するためです。プロセスの実行は、1 つのトランザクションの空間内で、ある "待機" 状態から別の "待機" 状態に移行します。

状況によっては、開発者はプロセス定義でトランザクションの境界を微調整する必要があります。jPDL では、プロセスの実行を非同期的に継続することを async="true" 属性で指定できます。async="true" は、イベントでトリガーされた場合にのみサポートされますが、すべてのノードタイプとすべてのアクションタイプで指定できます。

通常、ノードは、トークンがノードに進入した後に常に実行されます。したがって、ノードはクライアントのスレッドで実行されます。2 つの例を見て、非同期継続について説明します。最初の例は、3 つのノードを持つプロセスの一部です。ノード 'a' は待機状態、ノード 'b' は自動化されたステップ、ノード 'c' は待機状態です。このプロセスには非同期の動作は含まれません。以下にその図を示します。

最初のフレームは、開始状況を示しています。トークンはノード 'a' を指しています。これは、実行パスが外部のトリガーを待機していることを意味します。そのトリガーは、トークンにシグナルを送信することによって与える必要があります。シグナルが到達すると、トークンはノード 'a' からノード 'b' への遷移を介して渡されます。トークンがノード 'b' に到達すると、ノード 'b' が実行されます。ノード 'b' は、待機状態として動作しない自動化されたステップ (メールの送信など) であることに注意してください。そのため、2 番目のフレームは、ノード 'b' が実行されているときに取得したスナップショットです。ノード 'b' はプロセスの自動化されたステップであるため、ノード 'b' の実行には、ノード 'c' への遷移を介したトークンの伝播が含まれます。ノード 'c' は待機状態です。そのため、3 番目のフレームはシグナルメソッドが返された後の最終的な状況を示しています。

jBPM では "永続性" は必須ではありませんが、通常、シグナルはトランザクション内で呼び出されます。そのようなトランザクションの更新を見てみましょう。まず、トークンはノード 'c' を指すように更新されます。この更新は、JDBC 接続での GraphSession.saveProcessInstance の結果として、Hibernate によって生成されます。次に、自動化されたアクションが何らかのトランザクションリソースにアクセスして更新する場合、そのような更新は結合するか、同じトランザクションの一部にする必要があります。

2 番目の例は最初の例のバリアントであり、ノード 'b' に非同期継続が導入されています。ノード 'a' と 'c' は、最初の例と同じように動作します。つまり、待機状態として動作します。jPDL では、ノードは async="true" 属性を設定することによって非同期としてマークされます。

async="true" をノード 'b' に追加すると、プロセスの実行が 2 つの部分に分割されます。1 つ目の部分は、ノード 'b' が実行されるポイントまでプロセスを実行します。2 つ目の部分は、ノード 'b' を実行します。 その実行は待機状態 'c' で停止します。

したがって、トランザクションは、2 つの個別のトランザクション (各部分に 1 つずつ) に分割されます。1 つ目のトランザクションでノード 'a' から退出するには、外部のトリガー (Token.signal メソッドの呼び出し) が必要ですが、2 つ目のトランザクションは jBPM が自動的にトリガーして実行します。

アクションについても、原則は同様です。async="true" 属性でマークされたアクションは、プロセスを実行するスレッドの外部で実行されます。永続性が設定されている場合 (デフォルトの設定)、アクションは別のトランザクションで実行されます。

jBPM では、非同期メッセージングシステムを使用して非同期継続を実現します。プロセスの実行が非同期で実行されるべきポイントに到達すると、jBPM は実行を中断し、コマンドメッセージを生成してコマンドエグゼキューターに送信します。コマンドエグゼキューターは、メッセージを受信すると、停止されたプロセスの実行を再開する別のコンポーネントです。

jBPM は、JMS プロバイダーまたはそのビルトインの非同期メッセージングシステムを使用するように設定できます。ビルトインのメッセージングシステムは、機能面でかなり制限されていますが、JMS を使用できない環境でもこの機能をサポートできます。

ジョブエグゼキューター は、プロセスの実行を非同期的に再開するコンポーネントです。ジョブメッセージが非同期メッセージングシステムを介して到達するのを待機し、それらを実行します。非同期継続に使用される 2 つのジョブメッセージは、ExecuteNodeJob と ExecuteActionJob です。

これらのジョブメッセージは、プロセスの実行によって生成されます。プロセスの実行中、非同期で実行する必要があるノードまたはアクションごとに、Job (Plain Old Java Object) が MessageService に送信されます。メッセージサービスは、JbpmContext に関連付けられており、送信する必要があるすべてのメッセージを収集します。

メッセージは JbpmContext.close() の一部として送信されます。このメソッドは、close() の呼び出しを、関連するすべてのサービスに対してカスケードします。実際のサービスは jbpm.cfg.xml で設定できます。サービスの 1 つである JmsMessageService は、デフォルトで設定されており、新しいジョブメッセージが利用可能であることをジョブエグゼキューターに通知します。

グラフ実行メカニズムは、インターフェイス MessageServiceFactory および MessageService を使用してメッセージを送信します。これは、非同期メッセージングサービスを設定可能にするためです (これも jbpm.cfg.xml にあります)。Java EE 環境では、DbMessageService を JmsMessageService に置き換えて、アプリケーションサーバーの機能を活用できます。

以下は、ジョブエグゼキューターの仕組みの簡単な要約です。

"ジョブ" はデータベース内のレコードです。また、ジョブはオブジェクトであり、実行可能なものです。タイマーと非同期メッセージはどちらもジョブです。非同期メッセージの場合、dueDate は、メッセージが挿入されたときに現在の時刻に設定されます。ジョブエグゼキューターは、ジョブを実行する必要があります。これは 2 つのフェーズで行われます。

ジョブの取得とジョブの実行は、2 つの別個のトランザクションで行われます。ディスパッチャースレッドは、このノード上のすべてのエグゼキュータースレッドに代わって、データベースからジョブを取得します。エグゼキュータースレッドは、ジョブを受け取ると、その名前をジョブの所有者フィールドに追加します。各スレッドには、IP アドレスとシーケンス番号に基づく一意の名前があります。

ジョブの取得と実行の間に、スレッドが失敗することがあります。このような状況が発生した後にクリーンアップを行うために、ロック時間を確認するロックモニタースレッドがジョブエグゼキューターごとに 1 つあります。ロックモニタースレッドは、10 分以上ロックされていたジョブのロックを解除し、別のジョブ実行スレッドで実行できるようにします。

Hibernate の楽観的ロックを正しく機能させるためには、分離レベルを REPEATABLE_READ に設定する必要があります。REPEATABLE_READ に設定すると、競合するトランザクションのうちの 1 つに含まれる 1 つの行だけが、このクエリーによって確実に更新されます。

update JBPM_JOB job
set job.version = 2
    job.lockOwner = '192.168.1.3:2'
where 
    job.version = 1

update JBPM_JOB job
set job.version = 2
    job.lockOwner = '192.168.1.3:2'
where 
    job.version = 1

繰り返し不可能な読み取りに設定すると、異常が発生する可能性があります。トランザクションは、以前に読み取ったデータを再び読み取り、トランザクションの前回の読み取り以降にコミットされた別のトランザクションによってデータが変更されていることを検出します。

繰り返し不可能な読み取りは楽観的ロックの問題です。そのため、分離レベル READ_COMMITTED では、繰り返し不可能な読み取りが発生する可能性があるため、十分ではありません。そのため、複数のジョブ実行スレッドを設定する場合は、REPEATABLE_READ が必要です。

ジョブエグゼキューターに関連する設定プロパティーは次のとおりです。

jBPM のビルトインの非同期メッセージングを使用する場合、ジョブメッセージはデータベースに永続化されて送信されます。このメッセージの永続化は、jBPM プロセスの更新と同じトランザクションまたは JDBC 接続で行うことができます。

ジョブメッセージは JBPM_JOB テーブルに格納されます。

POJO コマンドエグゼキューター (org.jbpm.msg.command.CommandExecutor) は、データベーステーブルからメッセージを読み取り、実行します。POJO コマンドエグゼキューターの典型的なトランザクションは次のようになります。

コマンドメッセージの実行が失敗した場合、トランザクションはロールバックされます。その後、データベース内のメッセージにエラーメッセージを追加する新しいトランザクションが開始されます。コマンドエグゼキューターは、例外を含むメッセージをすべて除外します。

コマンドメッセージに例外を追加するトランザクションが失敗すると、ロールバックされます。メッセージは例外のない状態でキューに残り、後で再試行されます。