使用 boto3 时从 AWS SWF 接收重复的决策任务

Question

我创建了一个简单的 SWF 工作流程，但似乎收到了多个可用新决策任务的通知。我正在使用 boto3 python sdk。

我找不到好的boto3 swf示例代码，所以我从http://boto.cloudhackers.com/en/latest/swf_tut.html的boto2示例开始。

我已使用此脚本创建了我的简单工作流，该脚本在该工作流中创建了域、工作流和单个任务：

#!/usr/bin/python

import boto3
from botocore.exceptions import ClientError


swf = boto3.client('swf')

try:
  swf.register_domain(
    name="surroundiotest-swf",
    description="Surroundio test SWF domain",
    workflowExecutionRetentionPeriodInDays="10"
  )
except ClientError as e:
  print "Domain already exists: ", e.response.get("Error", {}).get("Code")

try:
  swf.register_workflow_type(
    domain="surroundiotest-swf",
    name="testflow",
    version="0.1",
    description="testworkflow",
    defaultExecutionStartToCloseTimeout="250",
    defaultTaskStartToCloseTimeout="NONE",
    defaultChildPolicy="TERMINATE",
    defaultTaskList={"name": "testflow"}
  )
  print "testflow created!"
except ClientError as e:
  print "Workflow already exists: ", e.response.get("Error", {}).get("Code")

try:
  swf.register_activity_type(
    domain="surroundiotest-swf",
    name="testworker",
    version="0.1",
    description="testworker",
    defaultTaskStartToCloseTimeout="NONE",
    defaultTaskList={"name": "testflow"}
  )
  print "testworker created!"
except ClientError as e:
  print "Activity already exists: ", e.response.get("Error", {}).get("Code")

我的工人代码：

#!/usr/bin/python

import boto3
from botocore.client import Config

botoConfig = Config(connect_timeout=50, read_timeout=70)
swf = boto3.client('swf', config=botoConfig)

print "Listening for Worker Tasks"

while True:

  task = swf.poll_for_activity_task(
    domain='surroundiotest-swf',
    taskList={'name': 'testflow'},
    identity='worker-1')

  if 'taskToken' not in task:
    print "Poll timed out, no new task.  Repoll"

  else:
    print "New task arrived"

    swf.respond_activity_task_completed(
        taskToken=task['taskToken'],
        result='success'
    )

    print "Task Done"

我的决策者代码：

#!/usr/bin/python

import boto3
from botocore.client import Config

botoConfig = Config(connect_timeout=50, read_timeout=70)
swf = boto3.client('swf', config=botoConfig)


print "Listening for Decision Tasks"

while True:

  newTask = swf.poll_for_decision_task(
    domain='surroundiotest-swf',
    taskList={'name': 'testflow'},
    identity='decider-1',
    reverseOrder=True)

  if 'taskToken' not in newTask:
    print "Poll timed out, no new task.  Repoll"

  elif 'events' in newTask:

    eventHistory = [evt for evt in newTask['events'] if not evt['eventType'].startswith('Decision')]
    lastEvent = eventHistory[-1]

    if lastEvent['eventType'] == 'WorkflowExecutionStarted':
      print "Dispatching task to worker", newTask['workflowExecution'], newTask['workflowType']
      swf.respond_decision_task_completed(
        taskToken=newTask['taskToken'],
        decisions=[
          {
            'decisionType': 'ScheduleActivityTask',
            'scheduleActivityTaskDecisionAttributes': {
                'activityType':{
                    'name': 'testworker',
                    'version': '0.1'
                    },
                'activityId': 'activityid-1001',
                'input': '',
                'scheduleToCloseTimeout': 'NONE',
                'scheduleToStartTimeout': 'NONE',
                'startToCloseTimeout': 'NONE',
                'heartbeatTimeout': 'NONE',
                'taskList': {'name': 'testflow'},
            }
          }
        ]
      )
      print "Task Dispatched"
      # print json.dumps(newTask, default=json_serial, sort_keys=True, indent=4, separators=(',', ': '))

    elif lastEvent['eventType'] == 'ActivityTaskCompleted':
      swf.respond_decision_task_completed(
        taskToken=newTask['taskToken'],
        decisions=[
          {
            'decisionType': 'CompleteWorkflowExecution',
            'completeWorkflowExecutionDecisionAttributes': {
              'result': 'success'
            }
          }
        ]
      )
      print "Task Completed!"

当我运行这些脚本并提交请求时，决策者接收任务并分派它，工作人员接收它并执行它。但是，决策者会在每次后续轮询中收到任务通知，因此我会看到决策者的输出如下：

Listening for Decision Tasks
Poll timed out, no new task.  Repoll
Dispatching task to worker {u'workflowId': u'surroundtest-1001', u'runId': u'23oPrHZ/d9kR43V/hr0ykZCI7Dks/FzLhfDeA9PPWFuPE='} {u'version': u'0.1', u'name': u'testflow'}
Task Dispatched
Dispatching task to worker {u'workflowId': u'surroundtest-1001', u'runId': u'23oPrHZ/d9kR43V/hr0ykZCI7Dks/FzLhfDeA9PPWFuPE='} {u'version': u'0.1', u'name': u'testflow'}
Task Dispatched
Dispatching task to worker {u'workflowId': u'surroundtest-1001', u'runId': u'23oPrHZ/d9kR43V/hr0ykZCI7Dks/FzLhfDeA9PPWFuPE='} {u'version': u'0.1', u'name': u'testflow'}
Task Dispatched
Dispatching task to worker {u'workflowId': u'surroundtest-1001', u'runId': u'23oPrHZ/d9kR43V/hr0ykZCI7Dks/FzLhfDeA9PPWFuPE='} {u'version': u'0.1', u'name': u'testflow'}
Task Dispatched
Dispatching task to worker {u'workflowId': u'surroundtest-1001', u'runId': u'23oPrHZ/d9kR43V/hr0ykZCI7Dks/FzLhfDeA9PPWFuPE='} {u'version': u'0.1', u'name': u'testflow'}
Task Dispatched

似乎 SWF 没有收到决策者已分派任务的通知，而是不断地将任务交付给决策者。是否有一些决策器设置我做错了，是否需要向 SWF 传达一些其他信息？

Answer 1

你有一个小错误。您正在使用reverseOrder=True 轮询决策任务。来自PollForDecisionTask的API文档：

reverseOrder
当设置为true 时，以相反的顺序返回事件。默认情况下，结果按事件的eventTimestamp 的升序返回。

使用reverseOrder=True，您最后得到最旧的事件，它始终是WorkflowExecutionStarted。任务完成后，你总是重新安排它。

通常，您希望使用previousStartedEventId 并仅处理在上一个决策任务之后发生的新事件。只处理最后一个活动任务并不总是足够的，因为可能需要处理多个事件。