示例方案

PDF

RSS

聚焦模式

示例方案 - AWS Flow Framework 适用于 Java

方案概述通用代码编写初始决策程序代码模拟不向后兼容的更改

可以将某些类型的代码更改视为不向后兼容。这些更改包括修改计划任务数量、类型或顺序的更新。考虑以下示例：

您编写决策程序代码以计划两个计时器任务。您启动一个执行并运行一个决策。因此，使用 IDs 1和调度了两个计时器任务2。

如果您更新决策程序代码以仅在要执行的下一个决策之前计划一个计时器，在下一个决策任务期间，该框架无法重播第二个 TimerFired 事件，因为 ID 2 与代码已生成的任何计时器任务不匹配。

方案概述

以下概述显示了该方案的步骤。该方案的最终目标是迁移到仅计划一个计时器的系统，但不会导致在迁移之前启动的执行失败。

初始决策程序版本
1. 编写决策程序。
2. 启动决策程序。
3. 决策程序计划两个计时器。
4. 决策程序启动 5 个执行。
5. 停止决策程序。
不向后兼容的决策程序更改
1. 修改决策程序。
2. 启动决策程序。
3. 决策程序计划一个计时器。
4. 决策程序启动 5 个执行。

以下几节包含说明如何实现该方案的 Java 代码示例。解决方案一节中的代码示例说明了几种修复不向后兼容的更改的方法。

注意

您可以使用最新版本的AWS SDK for Java运行该代码。

通用代码

不会在该方案中的示例之间更改以下 Java 代码。

SampleBase.java


package sample;
 
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
 
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.flow.JsonDataConverter;
import com.amazonaws.services.simpleworkflow.model.DescribeWorkflowExecutionRequest;
import com.amazonaws.services.simpleworkflow.model.DomainAlreadyExistsException;
import com.amazonaws.services.simpleworkflow.model.RegisterDomainRequest;
import com.amazonaws.services.simpleworkflow.model.Run;
import com.amazonaws.services.simpleworkflow.model.StartWorkflowExecutionRequest;
import com.amazonaws.services.simpleworkflow.model.TaskList;
import com.amazonaws.services.simpleworkflow.model.WorkflowExecution;
import com.amazonaws.services.simpleworkflow.model.WorkflowExecutionDetail;
import com.amazonaws.services.simpleworkflow.model.WorkflowType;
 
public class SampleBase {
 
    protected String domain = "DeciderChangeSample";
    protected String taskList = "DeciderChangeSample-" + UUID.randomUUID().toString();
    protected AmazonSimpleWorkflow service = AmazonSimpleWorkflowClientBuilder.defaultClient();
    {
        try {
            AmazonSimpleWorkflowClientBuilder.defaultClient().registerDomain(new RegisterDomainRequest().withName(domain).withDescription("desc").withWorkflowExecutionRetentionPeriodInDays("7"));
        } catch (DomainAlreadyExistsException e) {
        }
    }
 
    protected List<WorkflowExecution> workflowExecutions = new ArrayList<>();
 
    protected void startFiveExecutions(String workflow, String version, Object input) {
        for (int i = 0; i < 5; i++) {
            String id = UUID.randomUUID().toString();
            Run startWorkflowExecution = service.startWorkflowExecution(
                    new StartWorkflowExecutionRequest().withDomain(domain).withTaskList(new TaskList().withName(taskList)).withInput(new JsonDataConverter().toData(new Object[] { input })).withWorkflowId(id).withWorkflowType(new WorkflowType().withName(workflow).withVersion(version)));
            workflowExecutions.add(new WorkflowExecution().withWorkflowId(id).withRunId(startWorkflowExecution.getRunId()));
            sleep(1000);
        }
    }
 
    protected void printExecutionResults() {
        waitForExecutionsToClose();
        System.out.println("\nResults:");
        for (WorkflowExecution wid : workflowExecutions) {
            WorkflowExecutionDetail details = service.describeWorkflowExecution(new DescribeWorkflowExecutionRequest().withDomain(domain).withExecution(wid));
            System.out.println(wid.getWorkflowId() + " " + details.getExecutionInfo().getCloseStatus());
        }
    }
 
    protected void waitForExecutionsToClose() {
        loop: while (true) {
            for (WorkflowExecution wid : workflowExecutions) {
                WorkflowExecutionDetail details = service.describeWorkflowExecution(new DescribeWorkflowExecutionRequest().withDomain(domain).withExecution(wid));
                if ("OPEN".equals(details.getExecutionInfo().getExecutionStatus())) {
                    sleep(1000);
                    continue loop;
                }
            }
            return;
        }
    }
 
    protected void sleep(int millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
 
}

Input.java


package sample;
 
public class Input {
 
    private Boolean skipSecondTimer;
 
    public Input() {
    }
 
    public Input(Boolean skipSecondTimer) {
        this.skipSecondTimer = skipSecondTimer;
    }
 
    public Boolean getSkipSecondTimer() {
        return skipSecondTimer != null && skipSecondTimer;
    }
 
    public Input setSkipSecondTimer(Boolean skipSecondTimer) {
        this.skipSecondTimer = skipSecondTimer;
        return this;
    }
 
}

编写初始决策程序代码

下面是决策程序的初始 Java 代码。它注册为版本 1，并计划两个 5 秒计时器任务。

InitialDecider.java


package sample.v1;
 
import com.amazonaws.services.simpleworkflow.flow.DecisionContext;
import com.amazonaws.services.simpleworkflow.flow.DecisionContextProviderImpl;
import com.amazonaws.services.simpleworkflow.flow.WorkflowClock;
import com.amazonaws.services.simpleworkflow.flow.annotations.Execute;
import com.amazonaws.services.simpleworkflow.flow.annotations.Workflow;
import com.amazonaws.services.simpleworkflow.flow.annotations.WorkflowRegistrationOptions;
 
import sample.Input;
 
@Workflow
@WorkflowRegistrationOptions(defaultExecutionStartToCloseTimeoutSeconds = 60, defaultTaskStartToCloseTimeoutSeconds = 5)
public interface Foo {
 
    @Execute(version = "1")
    public void sample(Input input);
 
    public static class Impl implements Foo {
 
        private DecisionContext decisionContext = new DecisionContextProviderImpl().getDecisionContext();
        private WorkflowClock clock = decisionContext.getWorkflowClock();
 
        @Override
        public void sample(Input input) {
            System.out.println("Decision (V1) WorkflowId: " + decisionContext.getWorkflowContext().getWorkflowExecution().getWorkflowId());
            clock.createTimer(5);
            clock.createTimer(5);
        }
 
    }
}

模拟不向后兼容的更改

以下修改的决策程序 Java 代码是一个很好的不向后兼容更改例子。该代码仍注册为版本 1，但仅计划一个计时器。

ModifiedDecider.java


package sample.v1.modified;
 
import com.amazonaws.services.simpleworkflow.flow.DecisionContext;
import com.amazonaws.services.simpleworkflow.flow.DecisionContextProviderImpl;
import com.amazonaws.services.simpleworkflow.flow.WorkflowClock;
import com.amazonaws.services.simpleworkflow.flow.annotations.Execute;
import com.amazonaws.services.simpleworkflow.flow.annotations.Workflow;
import com.amazonaws.services.simpleworkflow.flow.annotations.WorkflowRegistrationOptions;
 
import sample.Input;
 
@Workflow
@WorkflowRegistrationOptions(defaultExecutionStartToCloseTimeoutSeconds = 60, defaultTaskStartToCloseTimeoutSeconds = 5)
public interface Foo {
 
    @Execute(version = "1")
    public void sample(Input input);
 
    public static class Impl implements Foo {
 
        private DecisionContext decisionContext = new DecisionContextProviderImpl().getDecisionContext();
        private WorkflowClock clock = decisionContext.getWorkflowClock();
 
        @Override
        public void sample(Input input) {
            System.out.println("Decision (V1 modified) WorkflowId: " + decisionContext.getWorkflowContext().getWorkflowExecution().getWorkflowId());
            clock.createTimer(5);
        }
 
    }
}

通过使用以下 Java 代码，您可以运行修改的决策程序以模拟进行不向后兼容更改的问题。

RunModifiedDecider.java


package sample;
 
import com.amazonaws.services.simpleworkflow.flow.WorkflowWorker;
 
public class BadChange extends SampleBase {
 
    public static void main(String[] args) throws Exception {
        new BadChange().run();
    }
 
    public void run() throws Exception {
        // Start the first version of the decider
        WorkflowWorker before = new WorkflowWorker(service, domain, taskList);
        before.addWorkflowImplementationType(sample.v1.Foo.Impl.class);
        before.start();
 
        // Start a few executions
        startFiveExecutions("Foo.sample", "1", new Input());
 
        // Stop the first decider worker and wait a few seconds 
        // for its pending pollers to match and return
        before.suspendPolling();
        sleep(2000);
 
        // At this point, three executions are still open, with more decisions to make
 
        // Start the modified version of the decider
        WorkflowWorker after = new WorkflowWorker(service, domain, taskList);
        after.addWorkflowImplementationType(sample.v1.modified.Foo.Impl.class);
        after.start();
 
        // Start a few more executions
        startFiveExecutions("Foo.sample", "1", new Input());
 
        printExecutionResults();
    }
 
}