Ich denke, Sie stellen einige sehr gute Fragen, die hervorheben, wie hilfreich SWF als Dienst sein kann. Kurz gesagt, Sie sagen Ihren Servern nicht, dass sie die Arbeit untereinander koordinieren sollen. Ihr Entscheidungsträger orchestriert all dies mithilfe des SWF-Dienstes für Sie.
Die Implementierung Ihres Workflows wird wie folgt ablaufen:
- Registrieren Ihres Workflows und seiner Aktivitäten beim Dienst (einmalig).
- Implementieren Sie den Entscheider und die Worker.
- Lassen Sie Ihre Arbeiter und Entscheider laufen.
- Starten Sie einen neuen Arbeitsablauf.
Es gibt eine Reihe von Möglichkeiten, Anmeldeinformationen in den Code von boto.swf einzufügen. Für diese Übung empfehle ich, sie in die Umgebung zu exportieren, bevor Sie den folgenden Code ausführen:
export AWS_ACCESS_KEY_ID=<your access key>
export AWS_SECRET_ACCESS_KEY=<your secret key>
1) Um die Domäne, den Workflow und die Aktivitäten zu registrieren, führen Sie Folgendes aus:
# 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) Entscheider und Arbeiter implementieren und ausführen.
# 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
Die Worker sind viel einfacher, Sie müssen keine Vererbung verwenden, wenn Sie dies nicht möchten.
# 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) Führen Sie Ihre Entscheider und Worker aus. Ihre Entscheider und Worker können von verschiedenen Hosts oder von ein und derselben Maschine ausgeführt werden. Öffnen Sie vier Terminals und führen Sie Ihre Schauspieler aus:
Zuerst Ihre Entscheidung
$ python -i ab_decider.py
>>> while ABDecider().run(): pass
...
Dann Worker A, Sie könnten dies von Server A aus tun:
$ python -i ab_workers.py
>>> while WorkerA().run(): pass
Dann Worker B, möglicherweise von Server B, aber wenn Sie sie alle von einem Laptop aus ausführen, funktioniert es genauso gut:
$ python -i ab_workers.py
>>> while WorkerB().run(): pass
...
4) Zum Schluss starten Sie den Workflow.
$ 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')
>>>
Wechseln Sie zurück, um zu sehen, was mit Ihren Schauspielern passiert. Sie werden möglicherweise nach einer Minute Inaktivität vom Dienst getrennt. Drücken Sie in diesem Fall die Aufwärtspfeiltaste + Eingabetaste, um erneut in die Abfrageschleife zu gelangen.
Sie können jetzt zum SWF-Panel Ihrer AWS-Verwaltungskonsole gehen, überprüfen, wie die Ausführungen ablaufen, und ihren Verlauf anzeigen. Alternativ können Sie es über die Befehlszeile abfragen.
>>> 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': ...
Dies ist nur ein Beispiel für einen Workflow mit serieller Ausführung von Aktivitäten, aber es ist dem Entscheider auch möglich, die parallele Ausführung von Aktivitäten zu planen und zu koordinieren.
Ich hoffe, das wird dir zumindest den Einstieg erleichtern. Für ein etwas komplexeres Beispiel eines seriellen Workflows empfehle ich, sich dies anzusehen.
Ich habe keinen Beispielcode zum Teilen, aber Sie können SWF definitiv verwenden, um die Ausführung von Skripts auf zwei Servern zu koordinieren. Die Hauptidee dabei ist, drei Codeteile zu erstellen, die mit SWF kommunizieren:
- Eine Komponente, die weiß, welches Skript zuerst ausgeführt werden soll und was zu tun ist, nachdem das erste Skript ausgeführt wurde. Dies wird in SWF-Begriffen als "Entscheider" bezeichnet.
- Zwei Komponenten, die jeweils verstehen, wie das spezifische Skript ausgeführt wird, das Sie auf jedem Computer ausführen möchten. Diese werden in SWF-Begriffen als "Activity Worker" bezeichnet.
Die erste Komponente, der Entscheider, ruft zwei SWF-APIs auf:PollForDecisionTask und RespondDecisionTaskCompleted. Die Polling-Anforderung gibt der Entscheidungskomponente den aktuellen Verlauf eines ausgeführten Workflows, im Grunde die "Wo bin ich"-Statusinformationen für Ihren Skript-Runner. Sie schreiben Code, der sich diese Ereignisse ansieht, und finden heraus, welches Skript ausgeführt werden soll. Diese "Befehle" zum Ausführen eines Skripts würden in Form einer Planung einer Aktivitätsaufgabe vorliegen, die als Teil des Aufrufs von RespondDecisionTaskCompleted zurückgegeben wird.
Die zweite Komponente, die Sie schreiben, die Activity Worker, ruft jeweils zwei SWF-APIs auf:PollForActivityTask und RespondActivityTaskCompleted. Die Polling-Anforderung gibt dem Activity Worker einen Hinweis darauf, dass er das ihm bekannte Skript ausführen soll, was SWF als Activity Task bezeichnet. Die von der Abfrageanforderung an SWF zurückgegebenen Informationen können einzelne ausführungsspezifische Daten enthalten, die als Teil der Planung der Aktivitätsaufgabe an SWF gesendet wurden. Jeder Ihrer Server würde SWF unabhängig nach Aktivitätsaufgaben abfragen, um die Ausführung des lokalen Skripts auf diesem Host anzuzeigen. Sobald der Worker das Skript ausgeführt hat, ruft er über die RespondActivityTaskCompleted-API zurück zu SWF.
Der Rückruf von Ihrem Activity Worker an SWF führt dazu, dass ein neuer Verlauf an die bereits erwähnte Entscheidungskomponente übergeben wird. Es sieht sich den Verlauf an, stellt fest, dass das erste Skript fertig ist, und plant die Ausführung des zweiten. Sobald es sieht, dass die zweite fertig ist, kann es den Workflow mit einer anderen Art von Entscheidung "schließen".
Sie starten den gesamten Prozess der Ausführung der Skripts auf jedem Host, indem Sie die StartWorkflowExecution-API aufrufen. Dadurch wird der Datensatz des gesamten Prozesses in SWF erstellt und der erste Verlauf an den Entscheiderprozess ausgegeben, um die Ausführung des ersten Skripts auf dem ersten Host zu planen.
Hoffentlich gibt dies etwas mehr Kontext dazu, wie diese Art von Workflow mit SWF durchgeführt werden kann. Wenn Sie es noch nicht getan haben, würde ich einen Blick auf die Entwickleranleitung auf der SWF-Seite werfen, um weitere Informationen zu erhalten.