간단한 Amazon SWF 애플리케이션 구축

Maven에서 새 프로젝트를 시작합니다.

mvn archetype:generate -DartifactId=helloswf \
-DgroupId=aws.example.helloswf -DinteractiveMode=false

그러면 표준 maven 프로젝트 구조의 새 프로젝트가 생성됩니다.

helloswf
├── pom.xml
└── src
    ├── main
    │   └── java
    │       └── aws
    │           └── example
    │               └── helloswf
    │                   └── App.java
    └── test
        └── ...

test 디렉터리와 그 안에 포함된 항목은 이 자습서에서 사용되지 않으므로 모두 무시하거나 삭제할 수 있습니다. 또한 App.java도 새 클래스로 대체될 것이므로 삭제할 수 있습니다.

프로젝트 pom.xml 파일을 편집하고 <dependencies> 블록 내에 해당 모듈에 대한 종속성을 추가하여 aws-java-sdk-simpleworkflow 모듈을 추가합니다.

<dependencies>
  <dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>aws-java-sdk-simpleworkflow</artifactId>
    <version>1.11.1000</version>
  </dependency>
</dependencies>

Maven에서 JDK 1.7+가 지원되는 프로젝트가 빌드되는지 확인합니다. pom.xml에서 프로젝트(<dependencies> 블록 앞 또는 뒤)에 다음을 추가합니다.

<build>
  <plugins>
    <plugin>
      <artifactId>maven-compiler-plugin</artifactId>
      <version>3.6.1</version>
      <configuration>
          <source>1.8</source>
          <target>1.8</target>
      </configuration>
    </plugin>
  </plugins>
</build>

프로젝트의 src/main/java/aws/example/helloswf/ 디렉터리에서 파일을 생성합니다.

package 선언을 각 파일의 시작 부분에 추가하여 네임스페이스를 선언합니다. 이 예제 프로젝트에는 다음이 사용됩니다.

package aws.example.helloswf;

AmazonSimpleWorkflowClient 클래스와 com.amazonaws.services.simpleworkflow.model 네임스페이스의 여러 클래스에 대한 import 선언을 추가합니다. 간단히 말해서, 다음을 사용하겠습니다.

import com.amazonaws.regions.Regions;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.model.*;

텍스트 편집기를 열고 HelloTypes.java 파일을 생성하여 공통 단계에 따라 패키지 선언과 가져오기를 추가합니다.

HelloTypes 클래스를 선언하고 등록된 활동 및 워크플로 유형에 사용할 값을 해당 클래스에 제공합니다.

    public static final String DOMAIN = "HelloDomain";
    public static final String TASKLIST = "HelloTasklist";
    public static final String WORKFLOW = "HelloWorkflow";
    public static final String WORKFLOW_VERSION = "1.0";
    public static final String ACTIVITY = "HelloActivity";
    public static final String ACTIVITY_VERSION = "1.0";

이러한 값은 코드 전체에서 사용됩니다.

String 선언 뒤에 AmazonSimpleWorkflowClient 클래스의 인스턴스를 생성합니다. 이 인스턴스는 AWS SDK for Java에서 제공하는 Amazon SWF 메서드의 기본 인터페이스입니다.

private static final AmazonSimpleWorkflow swf =
    AmazonSimpleWorkflowClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();

이전 코드 조각에서는 임시 자격 증명이 default 프로필과 연결되어 있다고 가정합니다. 다른 프로필을 사용하는 경우 위 코드를 다음과 같이 수정하고 profile_name을 실제 프로필 이름으로 바꾸십시오.

private static final AmazonSimpleWorkflow swf =
        AmazonSimpleWorkflowClientBuilder
                .standard()
                .withCredentials(new ProfileCredentialsProvider("profile_name"))
                .withRegion(Regions.DEFAULT_REGION)
                .build();

새 함수를 추가하여 SWF 도메인을 등록합니다. 도메인은 관련 SWF 활동 및 워크플로 유형을 포함할 논리적 컨테이너입니다. SWF 구성 요소는 동일한 도메인 안에 있는 경우에만 서로 통신할 수 있습니다.

    try {
        System.out.println("** Registering the domain '" + DOMAIN + "'.");
        swf.registerDomain(new RegisterDomainRequest()
            .withName(DOMAIN)
            .withWorkflowExecutionRetentionPeriodInDays("1"));
    } catch (DomainAlreadyExistsException e) {
        System.out.println("** Domain already exists!");
    }

도메인을 등록하는 경우 도메인에 이름(:, /, |, 제어 문자 또는 리터럴 문자열 '`arn'을 제외한 1~256자로 구성된 모든 문자 집합)과 보존 기간(Amazon SWF에서 워크플로 실행이 완료된 후 워크플로의 실행 내역 데이터를 보관하는 일수)을 지정합니다. 최대 워크플로 실행 보존 기간은 90일입니다. 자세한 내용은 RegisterDomainRequest를 참조하십시오.

해당 이름의 도메인이 이미 있으면 DomainAlreadyExistsException이 발생합니다. 도메인이 생성된 경우에는 관련이 없으므로 이 예외를 무시해도 됩니다.

함수를 추가하여 새 활동 유형을 등록합니다. 활동은 워크플로의 작업 단위를 나타냅니다.

    try {
        System.out.println("** Registering the activity type '" + ACTIVITY +
            "-" + ACTIVITY_VERSION + "'.");
        swf.registerActivityType(new RegisterActivityTypeRequest()
            .withDomain(DOMAIN)
            .withName(ACTIVITY)
            .withVersion(ACTIVITY_VERSION)
            .withDefaultTaskList(new TaskList().withName(TASKLIST))
            .withDefaultTaskScheduleToStartTimeout("30")
            .withDefaultTaskStartToCloseTimeout("600")
            .withDefaultTaskScheduleToCloseTimeout("630")
            .withDefaultTaskHeartbeatTimeout("10"));
    } catch (TypeAlreadyExistsException e) {
        System.out.println("** Activity type already exists!");
    }

활동 유형은 이름과 버전으로 식별되는데, 이러한 항목은 해당 활동을 등록된 도메인 내 다른 활동으로부터 고유하게 식별하는 데 사용됩니다. 활동에는 활동 실행의 다양한 부분을 수행하는 데 걸리는 시간에 대한 제약 조건을 설정할 때 사용할 수 있는 다양한 제한 시간이나 SWF에서 작업 및 데이터를 받는 데 사용되는 기본 작업 목록과 같은 선택적 파라미터가 포함됩니다. 자세한 내용은 RegisterActivityTypeRequest를 참조하십시오.

텍스트 편집기를 열고 ActivityWorker.java 파일을 생성하여 공통 단계에 따라 패키지 선언과 가져오기를 추가합니다.

import com.amazonaws.regions.Regions;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.model.*;

ActivityWorker 클래스를 파일에 추가하고 Amazon SWF와 상호 작용하는 데 사용할 SWF 클라이언트를 포함하는 데이터 멤버를 제공합니다.

    private static final AmazonSimpleWorkflow swf =
            AmazonSimpleWorkflowClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();

활동으로 사용할 메서드를 추가합니다.

private static String sayHello(String input) throws Throwable {
    return "Hello, " + input + "!";
}

이 활동은 단순히 문자열을 받아서 인사말로 결합하고 결과를 반환합니다. 이 활동은 예외가 발생할 가능성이 매우 희박하지만, 잘못되었을 경우 오류를 일으킬 수 있는 활동을 설계하는 것이 바람직합니다.

활동 작업 폴링 메서드로 사용할 main 메서드를 추가합니다. 활동 작업의 작업 목록을 폴링하기 위한 몇 가지 코드를 추가하는 것부터 시작하겠습니다.

        System.out.println("Polling for an activity task from the tasklist '"
                + HelloTypes.TASKLIST + "' in the domain '" +
                HelloTypes.DOMAIN + "'.");

        ActivityTask task = swf.pollForActivityTask(
            new PollForActivityTaskRequest()
                .withDomain(HelloTypes.DOMAIN)
                .withTaskList(
                    new TaskList().withName(HelloTypes.TASKLIST)));

        String task_token = task.getTaskToken();

활동은 SWF 클라이언트의 pollForActivityTask 메서드를 호출하고 전달된 PollForActivityTaskRequest에서 사용할 도메인과 작업 목록을 지정하여 Amazon SWF에서 작업을 받습니다.

작업이 수신되고 나면 작업의 getTaskToken 메서드를 호출하여 작업의 고유 식별자를 가져옵니다.

다음에는 들어오는 작업을 처리하는 코드를 작성합니다. 작업을 폴링하고 관련 작업 토큰을 가져오는 코드 바로 뒤에 다음을 main 메서드에 추가합니다.

    if (task_token != null) {
        String result = null;
        Throwable error = null;

        try {
            System.out.println("Executing the activity task with input '" +
                    task.getInput() + "'.");
            result = sayHello(task.getInput());
        } catch (Throwable th) {
            error = th;
        }

        if (error == null) {
            System.out.println("The activity task succeeded with result '"
                    + result + "'.");
            swf.respondActivityTaskCompleted(
                new RespondActivityTaskCompletedRequest()
                    .withTaskToken(task_token)
                    .withResult(result));
        } else {
            System.out.println("The activity task failed with the error '"
                    + error.getClass().getSimpleName() + "'.");
            swf.respondActivityTaskFailed(
                new RespondActivityTaskFailedRequest()
                    .withTaskToken(task_token)
                    .withReason(error.getClass().getSimpleName())
                    .withDetails(error.getMessage()));
        }
    }

작업 토큰이 null이 아니면 작업과 함께 전송된 입력 데이터를 지정하여 활동 메서드(sayHello)의 실행을 시작할 수 있습니다.

작업이 성공하면(오류가 생성되지 않은 경우) 작업자는 작업 토큰과 활동의 결과 데이터가 포함된 RespondActivityTaskCompletedRequest 객체를 사용하여 SWF 클라이언트의 respondActivityTaskCompleted 메서드를 호출하여 SWF에 응답합니다.

반면, 작업이 실패하면 RespondActivityTaskFailedRequest 객체를 사용하여 respondActivityTaskFailed 메서드를 호출하고 작업 토큰과 오류에 대한 정보를 전달하여 응답합니다.

텍스트 편집기를 열고 WorkflowWorker.java 파일을 생성하여 공통 단계에 따라 패키지 선언과 가져오기를 추가합니다.

파일에 몇 가지 추가적인 가져오기를 추가합니다.

import com.amazonaws.regions.Regions;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.model.*;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

WorkflowWorker 클래스를 선언하고 SWF 메서드에 액세스하는 데 사용되는 AmazonSimpleWorkflowClient 클래스의 인스턴스를 생성합니다.

    private static final AmazonSimpleWorkflow swf =
            AmazonSimpleWorkflowClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();

main 메서드를 추가합니다. 이 메서드는 SWF 클라이언트의 pollForDecisionTask 메서드를 사용하여 결정 작업을 폴링하며 계속 반복됩니다. PollForDecisionTaskRequest에서 세부 정보를 제공합니다.

    PollForDecisionTaskRequest task_request =
        new PollForDecisionTaskRequest()
            .withDomain(HelloTypes.DOMAIN)
            .withTaskList(new TaskList().withName(HelloTypes.TASKLIST));

    while (true) {
        System.out.println(
                "Polling for a decision task from the tasklist '" +
                HelloTypes.TASKLIST + "' in the domain '" +
                HelloTypes.DOMAIN + "'.");

        DecisionTask task = swf.pollForDecisionTask(task_request);

        String taskToken = task.getTaskToken();
        if (taskToken != null) {
            try {
                executeDecisionTask(taskToken, task.getEvents());
            } catch (Throwable th) {
                th.printStackTrace();
            }
        }
    }

작업이 수신되고 나면 getTaskToken 메서드를 호출합니다. 이 메서드는 작업을 식별하는 데 사용할 수 있는 문자열을 반환합니다. 반환된 토큰이 아닌 null 경우 executeDecisionTask 메서드에서 추가로 처리하여 작업 토큰과, 작업과 함께 전송된 HistoryEvent 객체 목록을 전달합니다.

작업 토큰(String)과 HistoryEvent 목록을 사용하는 executeDecisionTask 메서드를 추가합니다.

    List<Decision> decisions = new ArrayList<Decision>();
    String workflow_input = null;
    int scheduled_activities = 0;
    int open_activities = 0;
    boolean activity_completed = false;
    String result = null;

또한 다음과 같은 항목을 추적하기 위해 데이터 멤버를 설정할 수도 있습니다.

다음에는 executeDecisionTask 메서드에 의해 보고되는 이벤트 유형에 근거하여 작업과 함께 전송된 HistoryEvent 객체를 처리할 코드를 getEventType에 추가합니다.

System.out.println("Executing the decision task for the history events: [");
for (HistoryEvent event : events) {
    System.out.println("  " + event);
    switch(event.getEventType()) {
        case "WorkflowExecutionStarted":
            workflow_input =
                event.getWorkflowExecutionStartedEventAttributes()
                     .getInput();
            break;
        case "ActivityTaskScheduled":
            scheduled_activities++;
            break;
        case "ScheduleActivityTaskFailed":
            scheduled_activities--;
            break;
        case "ActivityTaskStarted":
            scheduled_activities--;
            open_activities++;
            break;
        case "ActivityTaskCompleted":
            open_activities--;
            activity_completed = true;
            result = event.getActivityTaskCompletedEventAttributes()
                          .getResult();
            break;
        case "ActivityTaskFailed":
            open_activities--;
            break;
        case "ActivityTaskTimedOut":
            open_activities--;
            break;
    }
}
System.out.println("]");

이 워크플로의 목적상, 우리에게 중요한 부분은 다음과 같습니다.

그 밖의 이벤트 유형은 워크플로에 필요할 경우 사용할 수 있습니다. 각 이벤트 유형에 대한 자세한 내용은 HistoryEvent 클래스 설명을 참조하십시오.

+ 참조: switch 문의 문자열은 Java 7에서 도입되었습니다. Java의 이전 버전을 사용 중인 경우 EventType 클래스를 사용하여 history_event.getType()에 의해 반환되는 String을 열거 값으로 변환했다가 필요에 따라 다시 String으로 변환할 수 있습니다.

switch 문 뒤에, 수신된 작업에 따라 해당하는 결정으로 응답하는 코드를 추가합니다.

if (activity_completed) {
    decisions.add(
        new Decision()
            .withDecisionType(DecisionType.CompleteWorkflowExecution)
            .withCompleteWorkflowExecutionDecisionAttributes(
                new CompleteWorkflowExecutionDecisionAttributes()
                    .withResult(result)));
} else {
    if (open_activities == 0 && scheduled_activities == 0) {

        ScheduleActivityTaskDecisionAttributes attrs =
            new ScheduleActivityTaskDecisionAttributes()
                .withActivityType(new ActivityType()
                    .withName(HelloTypes.ACTIVITY)
                    .withVersion(HelloTypes.ACTIVITY_VERSION))
                .withActivityId(UUID.randomUUID().toString())
                .withInput(workflow_input);

        decisions.add(
                new Decision()
                    .withDecisionType(DecisionType.ScheduleActivityTask)
                    .withScheduleActivityTaskDecisionAttributes(attrs));
    } else {
        // an instance of HelloActivity is already scheduled or running. Do nothing, another
        // task will be scheduled once the activity completes, fails or times out
    }
}

System.out.println("Exiting the decision task with the decisions " + decisions);

어느 경우든, 결정 정보는 메서드 맨 위에 선언된 Decision 목록에 추가됩니다.

작업을 처리하면서 수집된 Decision 객체의 목록을 반환하여 결정 작업을 완료합니다. 작성 중이었던 executeDecisionTask 메서드의 끝에 이 코드를 추가합니다.

swf.respondDecisionTaskCompleted(
    new RespondDecisionTaskCompletedRequest()
        .withTaskToken(taskToken)
        .withDecisions(decisions));

SWF 클라이언트의 respondDecisionTaskCompleted 메서드는 작업을 식별하는 작업 토큰과 Decision 객체의 목록을 사용합니다.

텍스트 편집기를 열고 WorkflowStarter.java 파일을 생성하여 공통 단계에 따라 패키지 선언과 가져오기를 추가합니다.

WorkflowStarter 클래스를 추가합니다.

package aws.example.helloswf;


import com.amazonaws.regions.Regions;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflow;
import com.amazonaws.services.simpleworkflow.AmazonSimpleWorkflowClientBuilder;
import com.amazonaws.services.simpleworkflow.model.*;

public class WorkflowStarter {
    private static final AmazonSimpleWorkflow swf =
            AmazonSimpleWorkflowClientBuilder.standard().withRegion(Regions.DEFAULT_REGION).build();
    public static final String WORKFLOW_EXECUTION = "HelloWorldWorkflowExecution";

    public static void main(String[] args) {
        String workflow_input = "{SWF}";
        if (args.length > 0) {
            workflow_input = args[0];
        }

        System.out.println("Starting the workflow execution '" + WORKFLOW_EXECUTION +
                "' with input '" + workflow_input + "'.");

        WorkflowType wf_type = new WorkflowType()
            .withName(HelloTypes.WORKFLOW)
            .withVersion(HelloTypes.WORKFLOW_VERSION);

        Run run = swf.startWorkflowExecution(new StartWorkflowExecutionRequest()
            .withDomain(HelloTypes.DOMAIN)
            .withWorkflowType(wf_type)
            .withWorkflowId(WORKFLOW_EXECUTION)
            .withInput(workflow_input)
            .withExecutionStartToCloseTimeout("90"));

        System.out.println("Workflow execution started with the run id '" +
                run.getRunId() + "'.");
    }
}

WorkflowStarter 클래스는 단일 메서드인 main으로 이루어지는데, 이 메서드는 워크플로의 입력 데이터로서 명령줄을 통해 전달되는 인수(선택 사항)를 사용합니다.

SWF 클라이언트 메서드인 startWorkflowExecution은 StartWorkflowExecutionRequest 객체를 입력으로 사용합니다. 여기서는 실행할 워크플로 유형과 도메인을 지정하는 것 외에, 다음을 제공합니다.

startWorkflowExecution가 반환하는 Run 객체는 워크플로 실행에 대한 Amazon SWF의 기록에서 해당 특정 워크플로 실행을 식별하는 데 사용할 수 있는 값인 실행 ID를 제공합니다.

+ 참조: 실행 ID는 Amazon SWF에서 생성하며 워크플로 실행 시작 시 전달하는 워크플로 실행 이름과 동일하지 않습니다.