버텍스 코어: 이벤트 버스(Event Bus)

참고 문서

• Eclipse: Vertx.io Core Manual(v.3.9.16)

이 문서의 내용

이벤트 버스(Event Bus)는 버텍스의 신경계에 해당합니다.

이벤트 버스는 여러 서버 노드에 분산 된 p2p(peer-to-peer) 메시징 시스템을 형성하여 서로 다른 버티클 간에 메시지를 주고 받을 수 있도록 도와줍니다.

모든 Vertx 객체는 하나의 이벤트 버스 객체를 갖고 있습니다. 이벤트 버스 객체에 접근하려면 eventBus() 메소드를 사용합니다.

EventBus eb = vertx.eventBus();

이벤트 버스 주소(Event Bus Address)

이벤트 버스를 통해 전달되는 메시지의 목적지는 이벤트 버스 주소입니다.주소는 단순 문자열(String)이며 어떠한 문자열을 사용하더라도 유효합니다(다만, 네임스페이스와 같은 규칙 체계를 정하여 사용하는 것이 현명합니다).

"request.create.an.apple"

메시지

이벤트 버스로 전달 할 수 있는 메시지는 Primitive, 문자열, 버퍼 그리고 JSON입니다.

버텍스에서는 메시징을 위해 JSON을 사용하는 것이 관행적이지만, 이벤트 버스는 유연하기 때문에 원하는 객체를 보내는 것도 가능합니다.

이벤트 버스로 사용자가 정의한 클래스의 객체를 전달하려면 코덱(codec)을 정의하고 등록합니다.

이벤트 버스로 전달되는 메시지 객체는 본문(Body)에 해당하며 헤더(Headers)를 추가하려면 DeliveryOptions를 사용합니다.

DeliveryOptions options = new DeliveryOptions();
options.addHeader("some-header", "some-value");
eventBus.send("news.uk.sport", "Yay! Someone kicked a ball", options);

메시지는 버티클의 송신자가 발송한 순서대로 버티클의 수신자에게 도달함을 보장합니다.

핸들러(Handler)

핸들러는 이벤트 버스로 전달된 메시지를 실행하는 함수입니다.

핸들러를 이벤트 버스 주소에 등록하면, 주소로 메시지가 전달되었을 때 함수가 호출됩니다.

여러 개의 핸들러가 동일한 주소 하나에 등록 될 수 있습니다.

반대로 하나의 핸들러가 여러 개의 주소에 등록 될 수도 있습니다.

핸들러는 consumer() 메소드를 사용하여 등록합니다.

EventBus eb = vertx.eventBus();

eb.consumer("news.uk.sport", message -> {
	System.out.println("I have received a message: " + message.body());
});

핸들러에 메시지가 도착하면 핸들러가 호출됩니다.

메시지는 MessageConsumer에 도달하여 핸들러에 전달됩니다.

이후 MessageConsumer 객체를 사용하여 핸들러를 취소하거나, 새로운 핸들러를 등록하거나 할 수 있습니다.

EventBus eb = vertx.eventBus();

MessageConsumer<String> consumer = eb.consumer("news.uk.sport");
consumer.handler(message -> {
	System.out.println("I have received a message: " + message.body());
});

핸들러는 주소에 등록될 때 클러스터링 환경에서 각 노드에 등록 완료되기까지 시간이 필요합니다.

completionHandler를 사용하면 노드에 등록이 완전히 끝났을 때 콜백 받을 수 있습니다.

consumer.completionHandler(res -> {
	if (res.succeeded()) {
		System.out.println("The handler registration has reached all nodes");
	} else {
		System.out.println("Registration failed!");
	}
});

핸들러를 등록 취소하려면 unregister() 메소드를 사용합니다.

consumer.unregister(res -> {
	if (res.succeeded()) {
		System.out.println("The handler un-registration has reached all nodes");
	} else {
		System.out.println("Un-registration failed!");
	}
});

구독-발행(Publish-Subscribe)

가장 대표적인 메시징 시스템 중 하나인 구독-발행(Pub/Sub, Publish-Subscribe)입니다.

이벤트 버스 객체에서 주소로 메시지를 발행(Publish)하면, 주소에 등록(Subscribe) 된 모든 핸들러 함수가 호출됩니다.

이벤트 버스 주소에 메시지를 발행하려면 publish() 메소드를 사용합니다.

eventBus.publish("news.uk.sport", "Yay! Someone kicked a ball");

Point-to-Point 및 요청-응답(Request-Response)

메시지는 주소에 등록된 하나의 핸들러 함수만 호출하는 Point-to-Point 메시징도 지원합니다.

만약 주소에 여러 개의 핸들러가 등록되었다면, 엄격하지 않은 라운드 로빈(non-strict round-robin) 알고리즘으로 핸들러가 선택됩니다.

이벤트 버스 주소에 메시지를 요청하려면 send() 메소드를 사용합니다.

eventBus.send("news.uk.sport", "Yay! Someone kicked a ball");

Point-to-Point 메시지에서 원한다면 요청에 대한 응답(Response)을 처리하는 핸들러를 등록 할 수 있습니다.

응답에 대한 핸들러는 메시지를 Consume하는 버티클이 요청을 정상적으로 수신하여 처리하였음을 의미합니다.

이를 위해 수신자는 컨텍스트(Context)에 대해서 reply() 메소드 또는 요청이 실패하였다면 fail() 메소드를 사용합니다.

MessageConsumer<String> consumer = eventBus.consumer("news.uk.sport");
consumer.handler(message -> {
	System.out.println("I have received a message: " + message.body());
	message.reply("how interesting!");
});

메시지를 송신했던 버티클은 다음과 같이 응답 핸들러를 등록합니다.

eventBus.request("news.uk.sport", "Yay! Someone kicked a ball across a patch of grass", ar -> {
	if (ar.succeeded()) {
		System.out.println("Received reply: " + ar.result().body());
	}
});

메시징 타임아웃

어떤 요청은 정해진 시간 내에 응답되지 않았을 때 타임아웃 처리 될 수 있습니다.

타임아웃 시간은 DeliveryOptions를 사용하여 지정 할 수 있습니다.

별도의 타임아웃 시간을 지정하지 않았다면 디폴트 타임아웃 시간은 30초입니다.

정리 및 복습

• 이벤트 버스를 통해 Pub/Sub와 Point-to-Point 메시징을 사용 할 수 있습니다.
• 이벤트 버스 주소는 이벤트 버스를 통해 메시지가 전달 되는 목적지입니다.
• 메시지는 Primitive 문자열 버퍼 JSON 객체가 될 수 있으며  코덱(codec)을 정의하고 등록하면 사용자 클래스 객체를 전달 할 수도 있습니다.
• 핸들러는 이벤트 버스 주소로 메시지가 전달되었을 때 호출되는 콜백 함수입니다.
• 하나의 핸들러가 여러 개의 이벤트 버스 주소에 등록되거나, 여러 개의 핸들러가 하나의 이벤트 버스 주소에 등록 될 수도 있습니다.
• 메시징의 디폴트 타임아웃은 30초이며 DeliveryOptions를 사용하여 별도 지정 할 수 있습니다.