저작권 안내: 저작권자표시 Yes 상업적이용 No 컨텐츠변경 No

스프링5 입문

JSP 2.3

JPA 입문

DDD Start

인프런 객체 지향 입문 강의

'supervisor'에 해당되는 글 2건

  1. 2014.01.28 CentOS에 Storm 설치해 보기
  2. 2011.03.22 Akka 세 번째, Supervisor를 이용한 Fault Tolerance

실시간 데이터 처리용으로 쓰임새가 늘어나고 있는 스톰Storm을 설치해 봤는데, 그 과정을 정리해 본다.


설치 순서

  1. 자바 설치 (1.6 이상 설치되어 있다고 가정)
  2. 주키퍼ZooKeeper 설치
  3. ZeroMQ 설치
  4. JZMQ 설치
  5. 스톰 님버스Nimbus 설정
  6. 스톰 수퍼바이저Supervisor 설정
  7. 클러스터 시작하기
본 설치 과정을 위해 CentOS 6버전이 설치된 5대의 장비를 사용했다. 각 호스트 이름 별로 다음의 데몬들을 설치할 것이다.
  • zk1: 주키퍼 설치
  • storm-nimbus: 님버스 설치, ZeroMQ, JZMQ
  • storm-sup1 ~ storm-sup3: 수퍼바이저 설치, ZeroMQ, JZMQ
참고로, 스톰의 님버스나 수퍼바이저가 뭔지 모른다면, 스톰 훑어보기 자료(http://javacan.tistory.com/324)를 참고하기 바란다.

주키퍼ZooKeeper 설치(zk1 장비)

1. 주키퍼 데이터 디렉토리 생성

이 글에서는 /data1/zk 를 생성한다고 가정

2. 원하는 디렉토리에 주키퍼 다운로드 받아 설치

여기서는 /data1/program 디렉토리에 설치한다고 가정. 아래 다운로드 경로는 http://www.apache.org/dyn/closer.cgi/zookeeper/ 에서 확인

$ cd /data1/program
$ wget http://mirror.apache-kr.org/zookeeper/stable/zookeeper-3.4.5.tar.gz
$ tar xvzf zookeeper-3.4.5.tar.gz

3. 주키퍼 설정 파일 작성

/data1/program/zookeeper-3.4.5/conf/zoo.cfg 파일 작성. 아래는 작성 예이다.

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data1/zk
clientPort=2181
autopurge.snapRetainCount=5
autopurge.purgeInterval=25


4. 주키퍼 실행


$ cd /data1/program/zookeeper-3.4.5/bin

./zkServer.sh start

JMX enabled by default

Using config: /data1/program/zookeeper-3.4.5/bin/../conf/zoo.cfg

Starting zookeeper ... STARTED



ZeroMQ 설치 (storm-nim, storm-sup1~storm-sup3 장비)


ZeroMQ는 스톰이 통신할 때 사용하는 라이브러리이다.


1. 필요 라이브러리 설치


$ yum install gcc gcc-c++ libuuid-devel


2. ZeroMQ 설치


$ wget http://download.zeromq.org/zeromq-4.0.3.tar.gz

$ tar xzf zeromq-4.0.3.tar.gz

$ cd zeromq-4.0.3

$ ./configure

$ make

$ make install


설치가 끝나면 /usr/local/lib/ 디렉토리에 libzmq.so 파일이 있는 확인해보자.


JZMQ 설치 (storm-nim, storm-sup1~storm-sup3 장비)


1. git 설치 / libtool 설치


git으로부터 소스를 다운로드 받으므로, git을 설치한다. git에서 clone 하기 싫다면, Github 사이트에서 직접 다운로드 받아도 된다.


$ yum install git libtool


2. JZMQ 설치


JAVA_HOME이 알맞게 설정되어 있어야 한다.


$ git clone https://github.com/nathanmarz/jzmq.git

$ cd jzmq

$ ./autogen.sh

$ ./configure

$ make

$ make install


/usr/local/share/java 디렉토리에 zmq.jar 파일이 생성된 것을 확인할 수 있다.



스톰 설치 (storm-nim, storm-sup1~storm-sup3 장비)


1. 스톰 다운로드


님버스와 수퍼바이저를 실행할 장비에 모두 Storm을 다운로드해서 설치한다. 아래 다운로드 주소는 http://storm-project.net/downloads.html 에서 구하면 된다.


$ wget https://dl.dropboxusercontent.com/s/tqdpoif32gufapo/storm-0.9.0.1.tar.gz

$ tar xzf storm-0.9.0.1.tar.gz

$ ln -s storm-0.9.0.1 storm


2. 스톰 데이터 디렉토리


님버스와 수퍼바이저로 사용될 장비에 스톰이 데이터를 보관할 디렉토리를 생성한다. 이 글에서는 /data1/storm 디렉토리를 사용한다고 가정한다.


3. 스톰 설정 파일 작성


각 장비의 [스톰설치경로]/conf/conf/storm.yaml 파일을 수정한다.


storm.zookeeper.servers:

     - "zk1"

 

nimbus.host: "storm-nim"


storm.local.dir: "/data1/storm"


ui.port: 8087


주키퍼 포트를 변경해야 한다거나, 디폴트 설정 값 등의 정보가 궁금하다면 https://github.com/nathanmarz/storm/blob/master/conf/defaults.yaml 를 참고하면 된다.


4. 스톰 클러스터 데몬 실행


storm-nim 장비에서 스톰 님버스 데몬을 실행한다. Ctrl+C 키를 누르면 데몬이 중지된다.


$ cd [스톰설치디렉토리]/bin

$ ./storm nimbus

Running: java -server -Dstorm.options= .......생략


storm-sup1 ~ storm-sup3 장비에서 동일한 방식으로 스톰 수퍼바이저 데몬을 실행한다. 마찬가지로 Ctrl+C 키를 누르면 중지된다.


$ cd [스톰설치디렉토리]/bin

$ ./storm supervisor

Running: java -server -Dstorm.options= .......생략


님버스 데몬을 실행 중인 장비에서 스톰 UI 웹 서버를 실행할 수 있다. 다음은 스톰 UI 데몬 실행 명령어이다.


$ ./storm ui


앞서 설정 파일에서 "ui.port"의 값을 8087로 주었는데, http://nimbus호스트:8087 로 연결해보면 다음과 비슷한 화면이 출력되는 것을 확인할 수 있다.



최초 환경 테스트를 수행할 때에는 위와 같이 스톰 데몬을 수동으로 띄우지만, 실제 환경에서는 supervisord나 daemontools, monit과 같은 툴을 이용해서 프로세스가 죽을 경우 자동으로 뜨도록 해 놓는다.


5. 간단한 명령어로 확인


아래 명령어로 클러스터에 등록된 토폴로지를 확인할 수 있다.


$ storm list

...생략

No topologies running.


설치가 완료되었다.




Posted by 최범균 madvirus

댓글을 달아 주세요

특정 노드에 갑자기 장애가 발생했다. 이럴 때 가장 먼저 개발자들이 선택하는 장애 대처 방법은? 아마도 관련 프로세스를 재시작하는 방법이 가장 많이 사용될 것이다. 웹서버를 재시작하거나 로그 수신기를 재시작하는 등 뭔가 문제가 발생한 영역의 프로세스나 쓰레드를 재시작함으로써 서비스 다운타임을 줄이는 것이 장애 발생 시 첫 번째로 수행하는 작업이다. 일단, 재시작해서 서비스가 살아나도록 만든 뒤, 그 다음에 원인 분석을 하게 된다.

Akka도 이와 비슷한 방법으로 액터의 장애에 대응할 수 있는 방법을 제공하고 있다. 액터는 자신을 관리하는 Supervisor를 가질 수 있으며, Supervisor는 액터가 다운될 경우 재시작함으로써 다운타임을 최소화하도록 해 준다. 이 방식은 얼랭(erlang)으로부터 빌려온 방식으로서 Akka는 Supervisor를 통해 무정지 서비스를 구현할 수 있도록 하고 있다.

액터의 두 가지 라이프 사이클: permanent, temporary

액터는 다음의 두 가지 라이프 사이클을 가진다.
  • permanent: 메시지 처리 과정에서 예외가 발생해도 액터가 살아 있음.
  • temporary: 메시지 처리 과정에서 예외가 발생하면 액터가 죽음.
액터를 permanent로 설정할 지 temporary로 설정할 지의 여부는 다음과 같이 설정할 수 있다.

import akka.config.Supervision;

public class WorkerActor extends UntypedActor {
    public WorkerActor() {
        getContext().setLifeCycle(Supervision.temporary());
        ...
    }


Akka의 액터 예외 대응 방식: Let it Crash

다중 쓰레드를 이용해서 병행 처리 코드를 작성할 경우, 병행 처리 코드에서 예외가 발생했을 때 이를 알 수 있는 방법은 예외 추적 메시지를 확인하는 방법 뿐이다. (또는 try-catch로 모든 예외를 잡아서 알림해 주는 기능을 넣는 방법 뿐이다.) 예외가 발생해서 병행 처리 쓰레드가 종료된 경우 이를 복구하는 방법은 재시작해주는 것 외에 특별한 방법이 없다.

Akka는 액터가 더 이상 정상적으로 동작할 수 없는 상태가 되어 메시지 처리 과정 중 예외를 발생시키면, 해당 액터를 재시작하는 방법으로 장애에 대응한다. 복구할 수 없는 예외 상황 발생시 액터가 뭔가 하지 않고 그냥 죽도록 놔두고 안정된 상태로 초기화하고 재시작하기 때문에, 이 방식을 "Let it Crash"라고 부른다.

수버바이저를 이용한 액터 관리

Akka는 수퍼바이저를 이용해서 액터를 관리한다. 수퍼바이저는 다른 액터를 모니터링하는 액터로서, 수퍼바이저 액터에서 다른 액터를 연결(link)함으로써 수퍼바이저가 다른 액터를 관리하게 된다.

수퍼바이저는 연결된 액터가 죽었을 때 다음의 두 가지 방식을 이용해서 연결된 액터를 재시작한다. 참고로 permanent 모드의 액터만 재시작되며, temporary 액터는 재시작되지 않는다.
  • One-For-One
  • All-For-One
One-For-One은 수퍼바이저와 연결된 액터가 죽으면, 죽은 액터만 재시작하고 나머지 연결된 액터는 그대로 유지한다. (아래 그림 참고)


[발췌: http://doc.akka.io/fault-tolerance-java]

반면 All-For-One은 수퍼바이저와 연결된 액터 중 하나가 죽으면, 연결된 모든 액터를 재시작한다. (아래 그림 참고) 이는 수퍼바이저에 의해 관리되는 액터 중 하나로도 비정상적으로 동작하면 나머지 액터들도 영향을 받아서 비정상적으로 동작하게 될 때에 사용된다.


[발췌: http://doc.akka.io/fault-tolerance-java]

수퍼바이저(Supervisor) 액터 만들기

수퍼바이저 액터는 일반 액터와 동일한 액터로서, 다음의 두 가지 방법을 이용해서 만들 수 있다.
  • link() 메서드를 이용
  • Supervisor 클래스를 이용해서 생성

link()를 이용한 액터 연결 및 관리

액터는 다른 액터를 연결함으로써 수퍼바이저 액터가 될 수 있으며, 연결할 때에는 link() 메서드를 사용한다. link()를 이용해서 액터를 관리할 경우 다음과 같은 방법으로 개발을 진행하면 된다.
  1. 수퍼바이저 액터로 동작할 클래스의 생성자에 FaultHandler를 지정한다. FaultHandler는 관리되는 액터가 죽었을 때, 그 액터만 재시작할 지 아니면 관리되는 모든 액터를 재시작할 지의 여부를 지정한다.
  2. 수퍼바이저 액터를 생성한 뒤, 관리할 액터를 link()로 연결한다.

1번, 수퍼바이저 액터를 직접 구현할 경우 다음과 같이 수퍼바이저 액터 생성자에서 재시작 전략을 지정해 주어야 한다.

import akka.actor.UntypedActor;
import akka.config.Supervision.OneForOneStrategy;

public class MasterActor extends UntypedActor {

    public MasterActor() {
        getContext().setFaultHandler(
                new OneForOneStrategy(
                        new Class[] { RuntimeException.class }, 3, 1000));
    }

    @Override
    public void onReceive(Object message) throws Exception {
        System.out.println("Master가 받은 메시지: " + message);
    }

}

위 코드에서 MasterActor는 관리하는 액터가 죽으면 해당 액터만 재시작하도록 설정하였다. OneForOneStrategy 객체를 생성할 때 첫 번째 파라미터는 액터를 재시작할 예외 타입을 지정한다. 위 코드는 모니터링 대상 액터의 onReceive() 메서드에서 RuntimeException이 발생하면 액터를 재시작한다는 것을 의미한다. 뒤의 두 숫자에 대해서는 뒤에서 다시 설명하겠다.

관리되는 액터가 죽을 때 관리되는 다른 액터들도 함께 재시작하고 싶은 경우에는 AllForOneStrategy 클래스를 사용하면 된다. 생성자에 전달되는 파라미터 목록은 OneForOneStrategy 클래스와 동일하다.

2번, 수퍼바이저 액터를 알맞게 구현했다면 그 다음으로 할 작업은 link() 메서드를 이용해서 수퍼바이저에 관리할 액터를 연결해 주는 것이다. 아래 코드는 예를 보여주고 있다.

ActorRef master = Actors.actorOf(MasterActor.class);
master.start();

ActorRef worker1 = Actors.actorOf(WorkerActor.class);
worker1.start();

master.link(worker1);

테스트를 위해 WorkerActor가 "die"라는 메시지를 받으면 RuntimeException을 발생시키도록 구현해 보았다.

@SuppressWarnings("unchecked")
public class WorkerActor extends UntypedActor {
    private static int num = 1;
   
    private int id;
    public WorkerActor() {
        id = num++;
        System.out.println("액터 생성됨: " + id);
    }
   
    @Override
    public void onReceive(Object message) throws Exception {
        if (message.equals("die")) {
            throw new RuntimeException("고의적 DIE");
        }
        System.out.println("Worker " + id + ": " + message);
    }
   
    @Override
    public void preRestart(Throwable cause) {
        System.out.println("Worker " + id + ": 재시작 전처리");
    }
   
    @Override
    public void postRestart(Throwable cause) {
        System.out.println("Worker " + id + ": 재시작 후처리");
    }
   
}

WorkerActor는 preRestart() 메서드와 postRestart() 메서드를 구현하고 있는데, 이 두 메서드는 각각 액터가 재시작하기 전/후에 호출된다. WorkerActor가 생성될 때 마다 1씩 증가된 id 값을 할당하는데 id 값을 새로 부여한 이유는 액터가 재시작할 때 액터 객체를 새로 생성하는 지의 여부를 확인하기 위해서다.

ActorRef master = Actors.actorOf(MasterActor.class);
master.start();

ActorRef worker1 = Actors.actorOf(WorkerActor.class);
worker1.start();
ActorRef worker2 = Actors.actorOf(WorkerActor.class);
worker2.start();

master.link(worker1); // master에 worker1 액터 연결
master.link(worker2); // master에 worker2 액터 연결

worker1.sendOneWay("메시지1-1");
worker2.sendOneWay("메시지2-1");
worker1.sendOneWay("메시지1-2");
worker2.sendOneWay("메시지2-2");

worker1.sendOneWay("die"); // worker1 액터 죽임!
worker1.sendOneWay("메시지1-3"); // worker1 액터에 메시지 전달
worker2.sendOneWay("메시지2-3");

위 코드는 중간에 worker1에 "die" 메시지를 보냄으로써 worker1을 죽인다. worker1 액터는 "die" 메시지를 받으면 RuntimeException을 발생시키는데, MasterWorker는 RuntimeException이 발생할 경우 해당 액터를 재시작하라고 설정하고 있다. 따라서, worker1 액터는 "die" 메시지를 받는 순간 RuntimeException을 발생시키며 죽지만 곧이어 재시작하게 되고, 따라서 죽은 이후에 받은 "메시지1-3" 메시지를 재시작한 액터가 처리하게 된다.

실제 위 코드의 실행 결과는 다음과 같다. (Akka가 출력하는 로그 메시지 중 중요한 것만 남기고 나머지는 생략하였다.)

액터 생성됨: 1
액터 생성됨: 2
16:43:40.843 [main] DEBUG akka.actor.Actor$ - Linking actor [Actor[tuto3.WorkerActor:46f67fb0-506a-11e0-a0e5-001d92ad4c1a]] to actor [Actor[tuto3.MasterActor:46f1c4c0-506a-11e0-a0e5-001d92ad4c1a]]
16:43:40.843 [main] DEBUG akka.actor.Actor$ - Linking actor [Actor[tuto3.WorkerActor:46f67fb1-506a-11e0-a0e5-001d92ad4c1a]] to actor [Actor[tuto3.MasterActor:46f1c4c0-506a-11e0-a0e5-001d92ad4c1a]]
Worker 1: 메시지1-1
Worker 1: 메시지1-2
16:43:40.875 [akka:event-driven:dispatcher:global-1] ERROR akka.actor.Actor$ - Exception when invoking
    actor [Actor[tuto3.WorkerActor:46f67fb0-506a-11e0-a0e5-001d92ad4c1a]]
    with message [die]
16:43:40.875 [akka:event-driven:dispatcher:global-1] ERROR akka.actor.Actor$ - Problem
java.lang.RuntimeException: 고의적 DIE
    at tuto3.WorkerActor.onReceive(WorkerActor.java:18) ~[classes/:na]
    ...
Worker 2: 메시지2-1
Worker 2: 메시지2-2
Worker 2: 메시지2-3
16:43:40.968 [akka:event-driven:dispatcher:global-3] INFO  akka.actor.Actor$ - Restarting actor [tuto3.WorkerActor] configured as PERMANENT.
16:43:40.968 [akka:event-driven:dispatcher:global-3] DEBUG akka.actor.Actor$ - Invoking 'preRestart' for failed actor instance [tuto3.WorkerActor].
Worker 1: 재시작 전처리
액터 생성됨: 3
16:43:40.968 [akka:event-driven:dispatcher:global-3] DEBUG akka.actor.Actor$ - Invoking 'postRestart' for new actor instance [tuto3.WorkerActor].
Worker 3: 재시작 후처리
16:43:40.968 [akka:event-driven:dispatcher:global-3] DEBUG akka.actor.Actor$ - Restart: true for [tuto3.WorkerActor].
16:43:40.968 [akka:event-driven:dispatcher:global-3] DEBUG a.d.Dispatchers$globalExecutorBasedEventDrivenDispatcher$ - Resuming 46f67fb0-506a-11e0-a0e5-001d92ad4c1a
16:43:40.984 [akka:event-driven:dispatcher:global-4] DEBUG akka.dispatch.MonitorableThread - Created thread akka:event-driven:dispatcher:global-4
Worker 3: 메시지1-3

위 실행 결과를 보면 다음의 사실을 확인할 수 있다.
  • 재시작 전처리는 1번 Worker가 수행한다.
  • 전처리 후, worker1에 해당하는 새로운 액터 객체를 생성한다. (액터 생성된: 3)
  • 재시작 후처리는 3번 Worker가 수행한다.
  • 이후 worker1은 3번 Worker와 연결되며, "메시지1-3" 메시지는 3번 Worker가 수행하게 된다.
즉, 액터가 죽으면 그 액터 객체를 재사용하는 것이 아니라 새로운 액터 객체를 생성하는 방법으로 재시작하는 것을 알 수 있다.

Supervisor 클래스를 이용한 수퍼바이저 액터 생성

수퍼바이저 액터에서 직접 관리할 액터를 생성하는 경우가 아니면 수퍼바이저 액터를 별도로 구현하기 보다는 Akka가 제공하는 Supervisor 클래스를 이용하는 것이 편리하다.


import akka.actor.Supervisor;
import akka.actor.SupervisorFactory;
import akka.config.Supervision;
import akka.config.Supervision.OneForOneStrategy;
import akka.config.Supervision.Supervise;
import akka.config.Supervision.SupervisorConfig;

...

ActorRef worker1 = Actors.actorOf(WorkerActor.class);
ActorRef worker2 = Actors.actorOf(WorkerActor.class);

Supervise[] supervises = new Supervise[2];
supervises[0] = new Supervise(worker1, Supervision.permanent());
supervises[1] = new Supervise(worker2, Supervision.permanent());

OneForOneStrategy strategy = new OneForOneStrategy(
        new Class[] {RuntimeException.class}, 3, 3000);
SupervisorConfig config = new SupervisorConfig(strategy, supervises);

Supervisor supervisor = new SupervisorFactory(config).newInstance();
// supervisor 생성 시점에서 내부적으로 생성한 SupervisorActor와 worker1가 worker2가 시작됨

worker1.sendOneWay("메시지");

SupervisorFactory를 통해서 Supervisor를 생성하면, 내부적으로 SupervisorActor 타입의 액터를 생성하고, 그 액터에 SupervisorConfig에 지정된 모든 액터를 연결(link)하고, 각 액터를 시작(start) 한다.

내부적으로 생성한 SupervisorActor에 접근하고 싶다면, 다음과 같이 supervisor() 메서드를 사용하면 된다.

Supervisor supervisor = ...;
ActorRef supervisorActor = supervisor.supervisor();


재시작 횟수 제한

OneForOneStrategy나 AllForOneStrategy를 생성할 때 두 번째/세 번째 파라미터는 각각 제한된 시간 내의 최대 재시작 시도 회수와 제한을 시간 의미한다.  예를 들어, 아래 코드는 1초 이내에 최대 3번의 재시작 시도를 시도한다는 것을 의미한다. (1초 안에 재시작을 3번까지 허용한다는 의미가 아니다.)

new OneForOneStrategy(new Class[] { RuntimeException.class }, 3, 1000);

1초 안에 액터 재시작을 3번 실패하면 (예를 들어, postRestart() 메서드에서 런타임 예외가 발생해서 실패), 해당 액터에 대해 재시작 시도를 하지 않으며 더 이상 액터를 사용할 수 없게 된다.

액터 생성/연결 한번에 하기

ActorRef는 액터를 생성하고 관리하기 위한 메서드를 제공하고 있으며, 이들 메서드는 다음과 같다.
  • ActorRef spawn(Class clazz): 액터를 생성하고 시작한다.
  • ActorRef spawnLink(Class clazz): 액터를 생성하고 시작하고, 연결한다.
  • ActorRef spawnRemote(Class clazz, String host, int port, long timeout): 리모트 액터를 생성하고 시작한다.
  • void startLink(ActorRef actor): 액터를 시작하고 연결한다.
위 메서드를 이용하면 다음과 같이 코드를 조금 더 간결하게 작성할 수 있다.

ActorRef master = Actors.actorOf(MasterActor.class);
master.start();
ActorRef worker1 = master.spawnLink(WorkerActor.class); // worker1 액터 시작/연결 됨

ActorRef worker2 = Actors.actorOf(WorkerActor.class);
master.startLink(worker2); // worker2 시작/연결 됨


참고자료


Posted by 최범균 madvirus

댓글을 달아 주세요