GNU/Linux >> Linux の 問題 >  >> Linux

Amazon SWF を使用してサーバー間で通信するには

SWF がサービスとしてどれほど役立つかを強調する、非常に良い質問をされていると思います。つまり、サーバー間で作業を調整するようにサーバーに指示しません。ディサイダーは、SWF サービスの助けを借りて、これらすべてを調整します。

ワークフローの実装は次のようになります:

<オール>
  • ワークフローとそのアクティビティをサービスに登録する (1 回限り)
  • ディサイダーとワーカーを実装する
  • ワーカーとディサイダーを実行させます。
  • 新しいワークフローを開始します。

    • 認証情報を boto.swf のコードにフィードする方法はいくつかあります。この演習では、以下のコードを実行する前に環境にエクスポートすることをお勧めします:

    export AWS_ACCESS_KEY_ID=<your access key>
export AWS_SECRET_ACCESS_KEY=<your secret key>

    1) ドメイン、ワークフロー、アクティビティを登録するには、以下を実行します:

    # ab_setup.py
import boto.swf.layer2 as swf

DOMAIN = 'stackoverflow'
ACTIVITY1 = 'ServerAActivity'
ACTIVITY2 = 'ServerBActivity'
VERSION = '1.0'

swf.Domain(name=DOMAIN).register()
swf.ActivityType(domain=DOMAIN, name=ACTIVITY1, version=VERSION, task_list='a_tasks').register()
swf.ActivityType(domain=DOMAIN, name=ACTIVITY2, version=VERSION, task_list='b_tasks').register()
swf.WorkflowType(domain=DOMAIN, name='MyWorkflow', version=VERSION, task_list='default_tasks').register()

    2) ディサイダーとワーカーを実装して実行する

    # ab_decider.py
import time
import boto.swf.layer2 as swf

DOMAIN = 'stackoverflow'
ACTIVITY1 = 'ServerAActivity'
ACTIVITY2 = 'ServerBActivity'
VERSION = '1.0'

class ABDecider(swf.Decider):

    domain = DOMAIN
    task_list = 'default_tasks'
    version = VERSION

    def run(self):
        history = self.poll()
        # Print history to familiarize yourself with its format.
        print history
        if 'events' in history:
            # Get a list of non-decision events to see what event came in last.
            workflow_events = [e for e in history['events']
                               if not e['eventType'].startswith('Decision')]
            decisions = swf.Layer1Decisions()
            # Record latest non-decision event.
            last_event = workflow_events[-1]
            last_event_type = last_event['eventType']
            if last_event_type == 'WorkflowExecutionStarted':
                # At the start, get the worker to fetch the first assignment.
                decisions.schedule_activity_task('%s-%i' % (ACTIVITY1, time.time()),
                   ACTIVITY1, VERSION, task_list='a_tasks')
            elif last_event_type == 'ActivityTaskCompleted':
                # Take decision based on the name of activity that has just completed.
                # 1) Get activity's event id.
                last_event_attrs = last_event['activityTaskCompletedEventAttributes']
                completed_activity_id = last_event_attrs['scheduledEventId'] - 1
                # 2) Extract its name.
                activity_data = history['events'][completed_activity_id]
                activity_attrs = activity_data['activityTaskScheduledEventAttributes']
                activity_name = activity_attrs['activityType']['name']
                # 3) Optionally, get the result from the activity.
                result = last_event['activityTaskCompletedEventAttributes'].get('result')

                # Take the decision.
                if activity_name == ACTIVITY1:
                    # Completed ACTIVITY1 just came in. Kick off ACTIVITY2.
                    decisions.schedule_activity_task('%s-%i' % (ACTIVITY2, time.time()),
                        ACTIVITY2, VERSION, task_list='b_tasks', input=result)
                elif activity_name == ACTIVITY2:
                    # Server B completed activity. We're done.
                    decisions.complete_workflow_execution()

            self.complete(decisions=decisions)
            return True

    ワーカーははるかに単純です。必要がなければ継承を使用する必要はありません。

    # ab_worker.py
import os
import time
import boto.swf.layer2 as swf

DOMAIN = 'stackoverflow'
ACTIVITY1 = 'ServerAActivity'
ACTIVITY2 = 'ServerBActivity'
VERSION = '1.0'

class MyBaseWorker(swf.ActivityWorker):

    domain = DOMAIN
    version = VERSION
    task_list = None

    def run(self):
        activity_task = self.poll()
        print activity_task
        if 'activityId' in activity_task:
            # Get input.
            # Get the method for the requested activity.
            try:
                self.activity(activity_task.get('input'))
            except Exception, error:
                self.fail(reason=str(error))
                raise error

            return True

    def activity(self, activity_input):
        raise NotImplementedError

class WorkerA(MyBaseWorker):
    task_list = 'a_tasks'

    def activity(self, activity_input):
        result = str(time.time())
        print 'worker a reporting time: %s' % result
        self.complete(result=result)

class WorkerB(MyBaseWorker):
    task_list = 'b_tasks'

    def activity(self, activity_input):
        result = str(os.getpid())
        print 'worker b returning pid: %s' % result
        self.complete(result=result)

    3) ディサイダーとワーカーを実行します。ディサイダーとワーカーは、別々のホストから実行することも、同じマシンから実行することもできます。 4 つのターミナルを開き、アクターを実行します。

    まずあなたの決定者

    $ python -i ab_decider.py 
>>> while ABDecider().run(): pass
...

    次に、ワーカー A は、サーバー A からこれを行うことができます:

    $ python -i ab_workers.py 
>>> while WorkerA().run(): pass

    次にワーカー B、おそらくサーバー B からですが、それらすべてをラップトップから実行すると、同様に機能します:

    $ python -i ab_workers.py 
>>> while WorkerB().run(): pass
...

    4) 最後に、ワークフローを開始します。

    $ python
Python 2.6.5 (r265:79063, Apr 16 2010, 13:57:41) 
[GCC 4.4.3] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import boto.swf.layer2 as swf
>>> workflows = swf.Domain(name='stackoverflow').workflows()
>>> workflows
[<WorkflowType 'MyWorkflow-1.0' at 0xdeb1d0>]
>>> execution = workflows[0].start(task_list='default_tasks')
>>>

    元に戻って、アクターに何が起こるかを確認してください。非アクティブ状態が 1 分間続くと、サービスから切断される場合があります。その場合は、上向き矢印 + Enter キーを押して、ポーリング ループに再度入ります。

    これで、AWS マネジメント コンソールの SWF パネルに移動して、実行状況を確認し、履歴を表示できます。または、コマンド ラインからクエリを実行することもできます。

    >>> execution.history()
[{'eventId': 1, 'eventType': 'WorkflowExecutionStarted', 
'workflowExecutionStartedEventAttributes': {'taskList': {'name': 'default_tasks'}, 
'parentInitiatedEventId': 0, 'taskStartToCloseTimeout': '300', 'childPolicy': 
'TERMINATE', 'executionStartToCloseTimeout': '3600', 'workflowType': {'version': 
'1.0', 'name': 'MyWorkflow'}}, 'eventTimestamp': 1361132267.5810001}, {'eventId': 2, 
'eventType': 'DecisionTaskScheduled', 'decisionTaskScheduledEventAttributes': 
{'startToCloseTimeout': '300', 'taskList': {'name': ...

    これは、アクティビティを順次実行するワークフローの単なる例ですが、ディサイダーがアクティビティの並列実行をスケジュールおよび調整することも可能です。

    これで少なくとも開始できることを願っています。シリアル ワークフローのもう少し複雑な例については、こちらを参照することをお勧めします。


    共有できるサンプル コードはありませんが、SWF を使用して 2 つのサーバー間でスクリプトの実行を調整することは間違いなく可能です。これに関する主なアイデアは、SWF と対話する 3 つのコードを作成することです:

    • 最初に実行するスクリプトと、最初のスクリプトの実行が完了したら何をすべきかを認識しているコンポーネント。これは、SWF 用語で「ディサイダー」と呼ばれます。
    • 各マシンで実行する特定のスクリプトの実行方法をそれぞれが理解する 2 つのコンポーネント。これらは、SWF 用語で「アクティビティ ワーカー」と呼ばれます。

    最初のコンポーネントであるディサイダーは、PollForDecisionTask と RespondDecisionTaskCompleted の 2 つの SWF API を呼び出します。ポーリング リクエストは、実行中のワークフローの現在の履歴、基本的にはスクリプト ランナーの「現在地」の状態情報をディサイダー コンポーネントに提供します。これらのイベントを見て、どのスクリプトを実行するかを判断するコードを記述します。スクリプトを実行するこれらの「コマンド」は、RespondDecisionTaskCompleted への呼び出しの一部として返されるアクティビティ タスクのスケジューリングの形式になります。

    作成する 2 番目のコンポーネントであるアクティビティ ワーカーは、それぞれ PollForActivityTask と RespondActivityTaskCompleted の 2 つの SWF API を呼び出します。ポーリング リクエストは、アクティビティ ワーカーに、SWF がアクティビティ タスクと呼ぶスクリプトを実行する必要があることを示します。ポーリング要求から SWF に返される情報には、アクティビティ タスクのスケジューリングの一部として SWF に送信された単一の実行固有のデータを含めることができます。各サーバーは、アクティビティ タスクの SWF を個別にポーリングして、そのホストでのローカル スクリプトの実行を示します。ワーカーがスクリプトの実行を完了すると、RespondActivityTaskCompleted API を介して SWF にコールバックします。

    アクティビティ ワーカーから SWF へのコールバックにより、既に説明したディサイダー コンポーネントに新しい履歴が渡されます。履歴を見て、最初のスクリプトが完了したことを確認し、2 番目のスクリプトの実行をスケジュールします。 2 つ目の処理が完了したことを確認すると、別の種類の決定を使用してワークフローを「閉じる」ことができます。

    StartWorkflowExecution API を呼び出して、各ホストでスクリプトを実行するプロセス全体を開始します。これにより、SWF でプロセス全体の記録が作成され、最初の履歴がディサイダー プロセスにキックされて、最初のホストでの最初のスクリプトの実行がスケジュールされます。

    これにより、SWF を使用してこのタイプのワークフローを実現する方法について、もう少し詳しい説明が得られることを願っています。まだご覧になっていない場合は、追加情報について SWF ページの開発者ガイドをご覧ください。


    Linux

    1. `sh` と `source` の使用の違いは何ですか?

    2. sed を使用して特定のパターンを置き換える

    3. Amazon EC2 インスタンス間の通信のベスト プラクティスは何ですか?

    1. SCPとFTPを使用してLinuxのサーバー間でファイルを転送する方法

    2. Imapsyncを使用してIMAPサーバー間でメールボックスを転送する方法

    3. 複数の Web サーバー間でアセットを共有する方法は?

    1. 2つのパターン間の（および除外する）行を印刷しますか？

    2. Ssh –保護されたファイルをサーバー間で1行にコピーしますか？

    3. 私が使用しているDnsサーバーは何ですか？