Taskurottaまたは分散システムでのプロセス制御

こんにちは、habrauser！



さまざまなサービスと既存のシステムを制御されたプロセスにリンクするタスクがあります。 速度は宇宙的ではなく必要です（つまり、株価に基づいて応答を作成することはできません）が、多くのプロセスがあり、使用する必要があるコンポーネント（システム）も迫っています。 p2pバインディングを行いたくありません。 美しくて管理しやすいものが欲しい。



市場を確認した後、Amazon Simple Workflowに基づいてレプリカを作成することが決定されました。これは、直接使用できないためです。 私たちに合ったフレームワークのプロパティ：

• 低開始しきい値（優れたプログラマーは今日高価です）。 ここでの低しきい値は、すべてが高レベルで（ほぼインターフェースとの対話レベルで）行われるため、プログラミングの開始という点でより重要です。 ただし、プロセスを非同期で適切に制御するには、もちろん経験を積む必要があります
• タスクのパラメーターと実行結果を保存するとき、プロセスタスクエグゼキューターの回帰テストの分析とベースを取得します
• 特定の場所（コーディネーター内）のプロセス制御ロジックの集中。 これは一見して明らかではないかもしれませんが、これは、各アクターが独自のロジックを持っている場合に考えられる代替案と比較して最大の利点です-他のコンポーネントが次に呼び出されます（転送制御）。 多くの場合、システムが複雑になり、コンポーネントを再利用できなくなります

これは私が望む最小値ですが、実践が示すように、より多くのプラスがあります。 このプロジェクトは「タスク」に敬意を表してタスクロッタと名付けられました。これはタスクであり、フィンランド語でgopherであり、重要ではありませんが表示されません。 オープンソースはGitHubで入手できます 。 プロジェクトはHazelcastを使用して実装され、サーバー間で共通のメモリ空間とランタイムを作成しました。RESTサービスを迅速かつ便利に実装するためのDropwizardと、最初の開発者であるAmazonの友人が初めて開発しました。 ドキュメントは難しいですが、すぐに修正します。



理論から実際の例に移りましょう。

ユーザーに文字列メッセージを送信するアプリケーションを開発する必要があるとします。 入力時に、ユーザーIDと文字セットを取得します。 彼のプロファイル（IDによる）から、電子メールまたは電話番号でメッセージを受信するための設定データを取得します。 電話番号とメールもプロファイルで入手できます。 次に、必要なトランスポートとともにメッセージを送信します。 送信が失敗した場合（アドレスまたは番号が正しくないため）、将来の再試行を防ぐためにプロファイルに記録する必要があります。

これに、メッセージを送信するサービスが既に存在し、他のサブネット上にあり、再利用する必要があるという非機能要件を追加します。



PS：説明されている例のすべてのソースコードは、GitHub taskurotta \ taskurotta-getstartedでも入手できます。





Taskurottaを使用すると、開発者にとって通常の方法で相互にやり取りするシステムコンポーネント（アクター）を実装できます-互いのメソッドを呼び出しますが、非同期で実行します。 アクターは、パフォーマーとコーディネーターの2つのタイプに分けられます。 実行者は、割り当てられたタスクを明確に遂行する必要があります。 それらは最も独立したモジュールであり、それに応じて-最も再利用可能です。 アーティストは、外部の世界（任意の入力/出力ストリーム）と対話し、必要な限り、そのような方法でタスクを実行できます。 一方、コーディネーターは外の世界に関連するタスクを実行しません。 データベース、ネットワーク、その他の不安定な可能性のあるコンポーネントとの直接的なやり取りにつまずかないように、できるだけ早く解決する必要があります。 実行者にタスクを設定し、アクションを調整し、それによってプロセスの実装（説明）を確実にする義務。 コーディネーターは、再利用可能なサブプロセスのパラダイムを実装することにより、他のコーディネーターのタスクを設定できます。



コーディネーターのタスクは、現在知られているタスクを配布することです。 つまり 結果を待ってブロックしないでください。 彼は、自分が知っているタスク間の依存関係を構築し、必要に応じて、さらなるアクションを決定するための非同期ポイントを形成する必要があります。



このプロセスでは、コーディネーターは次のことを行う必要があります。

1. ユーザープロフィールをリクエストする
2. プロフィールを待つ
3. ユーザーにメッセージを送信
4. 発送の結果を待つ

ステートマシン、複数のパフォーマーによる1つのメッセージの変更方法、その他の通常の骨の折れる方法を使用して、この一連のアクションをエンコードしません。 Promiseエンティティとコーディネーターのアクションを監視するシステムを使用して、簡単かつ美しく行います。

Promise<Profile> profilePromise = userProfileService.get(userId); Promise<Boolean> sendResultPromise = decider.sendToTransport(profilePromise, message);
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

この例は、サービスを呼び出した結果、実際のオブジェクトではなく、特定の約束、つまりタスクの結果へのリンクを取得することを示しています。 このPromiseを引数として他のサービス（タスク）に渡すことができます。 他のサービスの呼び出しはシステムによってインターセプトされ（つまり、実際の同期呼び出しは発生しません）、それらの間に関係が構築されます。 タイプPromiseのすべての引数が準備されるまで、タスクはサービスによって実行されません。 必要なすべての予備タスクが完了するまで。



したがって、プロセス制御はコーディネーターとシステムによって共同で実行されます。 コーディネーターはタスク間の依存関係を構築し、システムは、とりわけ、予備タスクの完了と、それに依存するタスクのその後の起動を待つ機能を引き継ぎます。



次に、さらなるアクションを決定するための非同期ポイントを明らかにします。

通知の送信が成功したことを確認する必要があり、そうでない場合は、ユーザーへの通知の送信をブロックする必要があります。

この場合、通知を送信した後、さらにアクションを実行する前に、送信結果を分析する必要があります。 つまり タスクが完了するまで待って分析し、結果に応じてブロックするかどうかを決定します。 このような問題を解決するために、コーディネーターは自分でタスクを作成する機会があります。 必要なPromiseの転送先となるさらなるアクションを定義するためのポイント。 以下のようになります。

  public void start(String userId, String message) { Promise<Profile> profilePromise = userProfileService.get(userId); Promise<Boolean> sendResultPromise = decider.sendToTransport(profilePromise, message); decider.blockOnFail(userId, sendResultPromise); } @Asynchronous public void blockOnFail(String userId, Promise<Boolean> sendResultPromise) { logger.info(".blockOnFail(userId = [{}], sendResultPromise = [{}])", userId, sendResultPromise); if (!sendResultPromise.get()) { userProfileService.blockNotification(userId); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

start（）メソッドはプロセスの開始です。 次は3つのタスクのステートメントです。 最初にプロファイルを受信し、2番目と3番目に、コーディネーターは結果の後続の分析（sendToTransportメソッドとblockOnFailメソッドの呼び出し）のために自分用に設定します。 したがって、コーディネーターは、最初のタスクの解決を待っているかのようになりますが、ブロックはしません。 問題が解決されると、TaskurottaシステムはsendToTransportコーディネーターメソッドを呼び出して、既製のPromiseオブジェクトを渡し、そこからget（）メソッドを使用して実際のデータを取得できます。 sendToTransportタスクが完了した後、blockOnFailタスクが起動され、通知の送信中にエラーが発生した場合にユーザーuserIdのメッセージをブロックするようにタスクをuserProfileServiceサービスに設定します。



さらにアクションを決定するためのポイントを使用して、さまざまなプロセス動作を実装できます。

• さまざまなブランチでの並列化
• Promise転送と@NoWaitアノテーションを使用して、独立したプロセススレッドを1ポイントでさらにマージする
• 非同期再帰
• 同様のタスクの実行の並列化。たとえば、すべてのファイルのデジタル署名を確認し、1つの決定点で実行結果を待つ
• など

PS：blockOnFailタスクは決定オブジェクトを介して呼び出されます。 これは、呼び出しをインターセプトする人工オブジェクトですが、実際にはblockOnFailメソッドを呼び出しません。 タスクを設定する必要があり、同期的に呼び出さないでください。



シナリオによると、電子メールとSMSを送信するための請負業者が既にいるため、プロファイルを操作する請負業者のみを作成できます。 この請負業者には2つのタスクがあります。

1. ユーザーIDでプロファイルを返す
2. 特定のユーザーにメッセージを送信できないことに関するプロファイルをプロファイルに作成します

インターフェースを宣言することから始めます。 コーディネーターはこのインターフェースで動作します。 以下、簡潔にするために、コードのコメントおよびその他の重要でない部分は省略されます。

  @Worker public interface UserProfileService { public Profile get(String userId); public void blockNotification(String userId); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

アノテーション@Worker



は、このインターフェイスを請負業者として定義します。 注釈には、その名前と（契約の）バージョンを定義するオプションの属性があります。 デフォルトでは、名前はインターフェースのフルネームであり、バージョンは「1.0」です。 異なるバージョンのエグゼキュータは、競合することなく異なるプロセスで同時に動作できます。



インターフェースの実装に移りましょう。

  public class UserProfileServiceImpl implements UserProfileService { private static final Logger logger = LoggerFactory.getLogger(UserProfileServiceImpl.class); @Override public Profile get(String userId) { return ProfileUtil.createRandomProfile(userId); } @Override public void blockNotification(String userId) { logger.info(".blockNotification(userId = [{}]", userId); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ここでは、プロファイルマネージャー（ProfileUtil）の初期化を省略しました。 データベース、LDAP、またはその他のレジストリで動作します。 この例は、請負業者がタスク（呼び出し）を受け取り、それらを実際のモジュールに委任することを示しています。



これで請負業者の作成が完了しました。



前に設定したタスクを解決するには、コーディネーターはまだ受信していないユーザープロファイル（Promiseオブジェクト）へのリンクを転送して、さらにアクションを決定する必要があります。 そこで、このユーザーへのメッセージ送信が既にブロックされている場合、彼はトランスポートを選択するか、何も送信しません。



ただし、エグゼキューターのインターフェイスは、エグゼキューター自身と同様に、結果を同期的に受信して提供するため、実行結果を宣言内のPromiseオブジェクトとして持たず、クリーンなデータオブジェクトを返します。 これは正しいです。 請負業者は、その使用方法を知る必要はありません。 たとえば、プロファイルを取得するための請負業者は、ユーザー識別子が既にわかっている場合、またはユーザーが不明で、この識別子をどこかから取得する別のタスクにリンクを転送する必要がある場合に使用できます。 したがって、請負業者とのやり取りのインターフェイスになります。 このインターフェースは、コーディネーターが必要に応じて決定します。 つまり コーディネーターのパッケージ（プロジェクト）で定義されます。 請負業者とのやり取りのインターフェイスを追加して、プロファイルを操作します。

  @WorkerClient(worker = UserProfileService.class) public interface UserProfileServiceClient { public Promise<Profile> get(String userId); public void blockNotification(String userId); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

@WorkerClient



アノテーションでマークされたインターフェイスが表示されます。 注釈パラメーターは、請負業者の実際のインターフェースのクラスを参照します。 このようにして、既存のインターフェースと特定のコーディネーターに必要なインターフェースとの間にリンクが確立されます。 このインターフェースを「請負業者のクライアントインターフェース」と呼びます。 このクライアントインターフェイスには、コーディネーターに必要なすべてのメソッド（未使用を宣言することはできません）および引数の同一の署名が含まれている必要があります。 まだ完了していないタスクの結果を引数として渡したい場合、任意の引数をPromiseタイプにすることができます。



それでは、最も興味深い部分、つまりコーディネーターの作成に移りましょう。 まず、コーディネーターインターフェイスを以下に示します。これを使用して、Taskurottaクライアントが必要なプロセスを開始します。

  @Decider public interface NotificationDecider { @Execute public void start(String userId, String message); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このインターフェースは@Decider



として定義されてい@Decider



-つまり コーディネーターとして。 この注釈には、 @Worker



注釈と同じプロパティ（名前とバージョン）があります。 デフォルトでは、インターフェースのフルネームが名前として使用され、「1.0」がバージョンとして使用されます。



startメソッドは@Execute



としてマークされ@Execute



。 これは、この方法でプロセスを開始できることを意味します。



次に、コーディネーターの実装に進みます

  public class NotificationDeciderImpl implements NotificationDecider { private static final Logger logger = LoggerFactory.getLogger(NotificationDeciderImpl.class); private UserProfileServiceClient userProfileService; private MailServiceClient mailService; private SMSServiceClient smsService; private NotificationDeciderImpl decider; @Override public void start(String userId, String message) { logger.info(".start(userId = [{}], message = [{}])", userId, message); Promise<Profile> profilePromise = userProfileService.get(userId); Promise<Boolean> sendResultPromise = decider.sendToTransport(profilePromise, message); decider.blockOnFail(userId, sendResultPromise); } @Asynchronous public Promise<Boolean> sendToTransport(Promise<Profile> profilePromise, String message) { logger.info(".sendToTransport(profilePromise = [{}], message = [{}])", profilePromise, message); Profile profile = profilePromise.get(); switch (profile.getDeliveryType()) { case SMS: { return smsService.send(profile.getPhone(), message); } case EMAIL: { return mailService.send(profile.getEmail(), message); } } return Promise.asPromise(Boolean.TRUE); } @Asynchronous public void blockOnFail(String userId, Promise<Boolean> sendResultPromise) { logger.info(".blockOnFail(userId = [{}], sendResultPromise = [{}])", userId, sendResultPromise); if (!sendResultPromise.get()) { userProfileService.blockNotification(userId); } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このコードでは、プライベートオブジェクトの初期化も省略しました。 完全で実用的なサンプルコードはtaskurotta-getstartedプロジェクトにあります。 ここで、プライベートフィールドの値は、コーディネーターのプロキシオブジェクトの特別なファクトリーを通じて取得されることに注意してください。



実装例では、コーディネーターによる未完了タスクの完了結果の待機ポイントが2つあります。 これは、sendToTransportおよびblockOnFailメソッドです。 これらのメソッドは、Promise型のすべての引数の準備ができたときにのみ呼び出されます。

つまり 対応するタスクが完了します。



タイプMailServiceClientおよびSMSServiceClientのフィールドオブジェクトは、対応する請負業者へのクライアントインターフェイスでもあります。 taskurotta-getstartedプロジェクトで初期化を確認することもできます。



現時点では、すべての請負業者とコーディネーターを実装しています。 アクター（エグゼキューターとコーディネーター）の立ち上げに直接進みます。



タスクは、アプリケーションサーバー内で実行することも、個別のJavaアプリケーションとして実行することもできます（この例では、taskurotta \ bootstrapモジュールとは別のアプリケーションのバリアントを使用します）。 別のアプリケーションは何をしますか：

• Taskurottaサーバーに登録済み
• Nスレッドのプールを開始してタスクを実行します
• Taskurottaサーバーからタスクを受信します
• 実行を開始します
• 結果をTaskurottaサーバーに転送します

別のJavaアプリケーションを実行するには、ブートストラップパッケージ、より具体的にはru.taskurotta.bootstrap.Mainクラスが使用されます。 YAML形式の設定ファイルの場所を引数として渡す必要があります。



どのように開始しようとしますか？ とても簡単です。 以下は、サーバー、アクター、およびソースコードからの起動の段階的なアセンブリです。 Linuxをお持ちでない場合は注意してください、マイナーな変更が必要です。



あなたはすでに持っていると仮定します：

• jdk 1.7
• メイヴン3
• git

サーバーTaskurottaを組み立てる

 git clone https://github.com/taskurotta/taskurotta.git cd taskurotta/
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

アセンブリを実行します。 スピードアップするには、テストを無効にします。

 mvn clean install -DskipTests
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

次に、2つのノードのクラスターを実行します（デモの目的で1台のマシンを使用しているため、起動オプションで異なるポートを使用しています）。 実際の環境では、必要な数のマシンを同じ構成で実行できます。



最初のクラスターノードを起動します。

 java -Xmx64m -Ddw.http.port=8081 -Ddw.http.adminPort=9081 -Ddw.logging.file.currentLogFilename="assemble/target/server1.log" -jar assemble/target/assemble-0.4.0-SNAPSHOT.jar server assemble/src/main/resources/hz.yml
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

2番目のノードを起動します（初期段階で意図的にメモリを制限してリークの可能性を特定します。この例の構成では、データベースは使用されないため、より多くのメモリを追加する必要があります）。

 java -Xmx64m -Ddw.http.port=8082 -Ddw.http.adminPort=9082 -Ddw.logging.file.currentLogFilename="assemble/target/server2.log" -jar assemble/target/assemble-0.4.0-SNAPSHOT.jar server assemble/src/main/resources/hz.yml
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

両方のサーバーが相互に接続すると、ログは次のメッセージのようになります。

  Members [2] { Member [192.168.1.2]:7777 Member [192.168.1.2]:7778 this }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ブラウザーでコンソールを開きます。 http：// localhost：8081 / index.html-最初のノードまたはhttp：// localhost：8082 / index.html -2番目のノード。



任意のノードを使用して、コンソールを操作できます。 主に同じ情報が表示されます。 現在、すべてのコンソール機能がこの構成で動作するわけではありません（データベースなし）。 すべてがoracleおよびmongodb dbの構成で機能します。 ドキュメントの展開オプションを参照してください。



それでは、プロセスを実行しましょう。 このためにtaskurotta-getstartedリポジトリを複製します

 git clone https://github.com/taskurotta/taskurotta-getstarted.git cd taskurotta-getstarted/ mvn clean install
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

アクターが作業を開始するには、プロセスを開始する必要があります。 たとえば91個を実行します。

 java -cp target/getstarted-process-1.0-SNAPSHOT.jar ru.taskurotta.example.starter.NotificationModule http://localhost:8081 91
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

コンソールhttp：// localhost：8081 / index.htmlを確認します。 [キュー]タブを選択します。 コーディネーター91には、91番目の実行プロセスに対応するタスクがあることがわかります。







次に、コーディネーターを実行します。 YAML構成ファイルで定義されているのは、エグゼキューターなしです。 したがって、起動後、プロセスのすべてのタスクが機能しなくなり、実行者のタスクが並んで表示されます。

 java -Xmx64m -jar target/getstarted-process-1.0-SNAPSHOT.jar -f src/main/resources/config-decider.yml
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

クラスターの最初のノードは、構成ファイルでコーディネーターのTaskurottaサーバーとして定義されています

  spreader: - Spreader: class: ru.taskurotta.example.bootstrap.SimpleSpreaderConfig instance: endpoint: "http://localhost:8081" threadPoolSize: 10 readTimeout: 0 connectTimeout: 3000
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

コンソールでキューリストを更新し、エグゼキューターを待っているタスクがあることを確認できます。



ここで、エグゼキュータを実行し（コーディネータを動作させます）、それらを2番目のクラスタノードに向けてデモンストレーションします。 クラスターノードは共有メモリと内部タスクを実行するための環境を形成するため、エグゼキューターからの要求がどのサーバーからのものであるかは関係ありません。

 java -Xmx64m -jar target/getstarted-process-1.0-SNAPSHOT.jar -f src/main/resources/config-workers.yml
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

パフォーマーには、相互作用のための2番目のクラスターノードがあります。

  spreader: - Spreader: class: ru.taskurotta.example.bootstrap.SimpleSpreaderConfig instance: endpoint: "http://localhost:8082" threadPoolSize: 10 readTimeout: 0 connectTimeout: 3000
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

その結果、すべてのプロセスが完全に機能し、管理コンソールで順番に確認できます。







これまでのところ、私が現時点で共有したいすべてのこと。 提案や建設的な批判を歓迎します。