在 AWS SWF 中运行并行活动

Question

我正在编写一个 AWS SWF 工作流程，其中第一个活动评估分区的数量并将其传递给子工作流程。决策者中的子工作流循环第一个活动返回的分区并启动最大值。允许并行活动。假设如果最大。允许的值为 50，一次并行启动 50 个活动。但是，我们面临的问题是直到所有 50 个完成后才开始下一个 50，即它停止所有其他分区直到所有 50 个完成。执行分 50 批执行。下面是代码示例：

@Override
public void execute(@Nonnull final Execution execution, @Nonnull final Step step)
        throws ExecutionException {
    Promise<Queue<Step>> stepQueuePromise = activitiesClient.partition(step, context);
    Promise<Void> exitCode = executeSteps(stepQueuePromise, execution, context);
    activitiesClient.verify(step, exitCode);
}

@Asynchronous
private Promise<Void> executeStepsInBatches(@Nonnull final Queue<Step> stepQueue,
                                   @Wait final List<Promise<Void>> previousSteps,
                                   @Nonnull final Execution execution,
                                   @Nonnull final Context context) {

    List<Promise<Void>> stepListPromises = new ArrayList<>();

    for (int i = 0; i < concurrentStepThreshold && !stepQueue.isEmpty(); i++) {
        Promise<Void> stepPromise = activitiesClient.execute(execution, stepQueue.poll(), context);
        stepListPromises.add(stepPromise);
    }

    if (!stepListPromises.isEmpty()) {
        return executeStepsInBatches(stepQueue, stepListPromises, execution, context);
    } else {
        return Promise.Void();
    }
}

我们希望批量执行 50 个活动，但一旦少数活动完成，应提交新活动以匹配 50 个并行运行计数。有人可以建议我们如何实现这一目标吗？

编辑（新代码）

我试过下面的代码：

@Override
public void execute(@Nonnull final Execution execution, @Nonnull final Step step)
        throws ExecutionException {

    Promise<Queue<Step>> stepQueuePromise = activitiesClient.partition(step, context);
    executeSteps(stepQueuePromise, execution, context);
}

@Asynchronous
private void executeSteps(@Nonnull final Promise<Queue<Step>> stepQueuePromise,
                          @Nonnull final Execution execution,
                          @Nonnull final Context context) {
    Integer numNotReady = 0;
    List<Promise<Void>> currentPromises = new ArrayList<>();
    Iterator<Step> inputItr = stepQueuePromise.get().iterator();
    while (inputItr.hasNext() && numNotReady < 20) {
        Promise<Void> promise = activitiesClient.execute(execution, inputItr.next(), context);
        currentPromises.add(promise);

        if (!promise.isReady()) {
            numNotReady++;
        }
    }
    log.info("Num of not ready" + numNotReady);
    waitForPromises(currentPromises);
}

@Asynchronous
void waitForPromises(@Wait final List<Promise<Void>> activityOutputs) {
}

第一个循环并行启动了 20 个活动。但是，即使决策者运行了新的活动，也没有提交。我可以看到我添加以验证决策程序运行的日志： 2018 年 8 月 6 日 17:16:34,962 [INFO] (SWF Decider ExecutorTaskList_1.0 1) com.amazon.traffic.cafe.orchestrator.swf.exec.impl.SwfExecutorImpl：未准备好的数量20 2018 年 8 月 6 日 17:16:50,808 [INFO] (SWF Decider ExecutorTaskList_1.0 1) com.amazon.traffic.cafe.orchestrator.swf.exec.impl.SwfExecutorImpl：

的数量

Answer 1

最后，下面的代码运行并测试：

@Override
public void execute(@Nonnull final Execution execution, @Nonnull final Step step)
        throws ExecutionException {


    Context context = getContext(execution, step);

    Promise<Queue<Step>> stepQueue = activitiesClient.partition(step, context);

    /**
     * List to hold the promise of started activities.
     */
    Promise<?>[] batchPromises = new Promise<?>[50];

    /**
     * Initialize the list with ready promises.
     */
    Arrays.fill(batchPromises, Promise.Void());

    /**
     * OrPromise list to unblock as soon as one of the activity is completed.
     */
    OrPromise waitForAtleastOneInBatch = new OrPromise(batchPromises);

    Promise<Void> exitCode = startActivityInBatch(execution, context, stepQueue, waitForAtleastOneInBatch);

}

@Asynchronous
private Promise<Void> startActivityInBatch(final Execution execution, final Context context,
                                            final Promise<Queue<Step>> stepQueue,
                                            final OrPromise waitForAtleastOneInBatch) {
    /**
     * Executes only when one of the promise is ready.
     */
    Promise<?>[] existingBatchPromises = waitForAtleastOneInBatch.getValues();

    /**
     * In this loop, we iterate over the promise list and if the promise is ready we replace it with
     * new promise by starting new activity.
     */
    for (int existingBatchIterator = 0; existingBatchIterator < existingBatchPromises.length;
         existingBatchIterator++) {
        /**
         * If the existing promise is ready, call the next task replace the ready promise.
         */
        if (existingBatchPromises[existingBatchIterator].isReady()) {
            final Step step = stepQueue.get().poll();
            if (step == null) {
                /**
                 * This means that queue is empty and we have run all the activities.
                 */
                existingBatchPromises[existingBatchIterator] = Promise.Void();
            } else {
                existingBatchPromises[existingBatchIterator] = activitiesClient.execute(execution, step, context);
            }
        }
    }

    /**
     * call recursively till  we have messages in queue.
     */
    if (stepQueue.get().size() > 0) {
        return startActivityInBatch(execution, context, stepQueue, new OrPromise(existingBatchPromises));
    } else {
        /**
         * AndPromise is used to make the workflow wait till all Promises are ready.
         */
        return new AndPromise(existingBatchPromises);
    }
}

Answer 2

我相信以下应该有效：

@Asynchronous
private Promise<Void> executeStepsInBatches(@Nonnull final Queue<Step> stepQueue,
                                   @Wait final List<Promise<Void>> previousSteps,
                                   @Nonnull final Execution execution,
                                   @Nonnull final Context context) {

    List<Promise<Void>> stepListPromises = new ArrayList<>();

    for (int i = 0; i < concurrentStepThreshold; i++) {
        Promise<Void> stepPromise = executeNext(stepQueue, execution, context, stepPromise);
        stepListPromises.add(stepPromise);
    }
    return Promises.listOfPromisesToPromise(stepListPromises);
}

@Asynchronous
private Promise<Void> executeNext(Queue<Step> stepQueue, Execution execution, Context context, Promise<?> previous) {
   if (stepQueue.isEmpty()) {
     return Promise.Void();
   }
   Promise<Void> stepPromise = activitiesClient.execute(execution, stepQueue.poll(), context);
   // Loop recursively
   return executeNext(stepQueue, execution, stepPromise);    
}