使用 Amazon SWF 在服务器之间传递消息？

1. 我想在服务器 A 上运行脚本 A
2. 完成后，我想向服务器 B 发送消息以运行脚本 B
3. 如果它成功完成，我希望它从工作流队列中清除作业

我真的很难弄清楚如何结合使用 Boto 和 SWF 来做到这一点。我不是在寻找一些完整的代码，而是在寻找是否有人可以对所涉及的内容进行更多解释。

• 我实际上如何告诉服务器 B 检查脚本的完成 一个？
• 如何确保服务器 A 不会获取脚本的完成 A 并尝试运行脚本 B(因为服务器 B 应该运行它)？
• 我如何实际通知 SWF 脚本 A 已完成？你是旗帜，还是 消息，还是什么？

我对这一切感到很困惑。我应该使用什么设计？

最佳答案

我认为您提出了一些很好的问题，突出了 SWF 作为一项服务的作用。简而言之，您不会告诉您的服务器在它们之间协调工作。在 SWF 服务的帮助下，您的决策者会为您安排所有这一切。

您的工作流程的实现如下:

1. 向服务注册您的工作流程及其事件(一次性)。
2. 实现决策者和 worker 。
3. 让您的工作人员和决策者运行。
4. 开始新的工作流程。

有多种方法可以将凭据提供给 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) 实现和运行决策者和 worker 。

# 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

worker就简单多了，不想用继承也可以不用。

# 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) 运行您的决策者和工作人员。 您的决策者和工作人员可能在不同的主机上运行，​​或者在一台和同一台机器上运行。打开四个终端并运行你的 Actor :

首先是你的决定者

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

那么 worker A，你可以从服务器 A 这样做:

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

然后是 worker 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')
>>> 

切换回去看看你的 Actor 会发生什么。他们可能会在一分钟不活动后断开与服务的连接。如果发生这种情况，请按向上箭头+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': ...

这只是一个连续执行事件的工作流示例，但决策者也有可能 schedule and coordinate parallel execution of activities .

我希望这至少能让您入门。对于一个稍微复杂的串行工作流示例，我推荐 looking at this .

关于python - 使用 Amazon SWF 在服务器之间进行通信，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/14829562/