Zk

ZooKeeper에 편하게 접속하기 위한 유틸리티입니다. 내부적으로 Apache Curator를 사용합니다.

접속 만들기

다음과 같이 ZooKeeper quorum의 주소를 입력하는 것으로 주키퍼 연결을 생성할 수 있습니다.

val zk = ZooKeeperConnection("quorum1.server.com,quorum2.server.com,quorum3.server.com")

설정 파일로 주키퍼 quorum을 관리할 수도 있습니다.

# application.conf
zookeeper {
  connectString = "quorum1.server.com,quorum2.server.com,quorum3.server.com"
  sessionTimeout = 6000
  connectionTimeout = 6000
}

# scala file
val zk = ZooKeeperConnection() 

노드 다루기

다음과 같이 주키퍼의 특정 노드에 대한 작업을 할 수 있는 ZNode를 받아올 수 있습니다.

val node = zk("path", "to", "node")

node.create("data")           // 노드 생성
node.set("data")              // 노드 값 설정
node.get().foreach(println)   // 노드 값 출력
node.delete()                 // 노드 삭제
node.deleteOrIgnore()         // 노드 삭제 - 이미 존재하지 않으면 무시
node.createOrSet("data")      // 노드 생성 - 이미 존재하면 값 설정
node.setOrCreate("data")      // 노드 값 설정 - 존재하지 않으면 생성
node.getChildren()            // 하위 노드 가져오기

Watch 기능 활용

Watch 기능을 사용하여 노드의 값이 변화하거나 하위 노드 목록이 변화했을 때 알림을 받을 수 있습니다. 콜백을 사용할 수도 있고 Rx Observable을 이용할 수도 있습니다.

# 콜백
node.watch {
  case Success(value) => println(s"New Value: $value")
  case Failure(error) => println(s"Error: $error")
}

# Observable로 higher-order function 사용하기
node.watch().map(_.length).subscribe {
  value => println(s"New Value's Length: $value")
}

# Observable 에러 처리
node.watch().subscribe (
  value => println(s"New Value: $value"),
  error => println(s"Error: $error");
)

# Unsubscribe
val subscription = node.watch { 
  case Success(value) => println(s"New Value: $println")
  case Failure(error) => println(s"Error: $error")
}
// 또는
val subscription = node.watch().subscribe (
  value => println(s"New Value: $println"),
  error => println(s"Error: $error")
)
// ...
subscription.unsubscribe()  // 더 이상 알림을 받지 않음