[Spring] Amazon MQ (Active MQ) : Consumer 운영

ActiveMQ를 적용해 보니 서비스 메시지 성향에 따라 Consumer 운영이 달라져야 한다고 생각됐다.

Message 간의 디펜던시 및 영향도가 있나 없나에 따른 운영 전략이 달라진다.

 

Message A : 메시지간의 영향이 없는 메시지

예를 들어 신규로 등록해야 할 상품 정보가 담긴 메시지가 Producer로부터 발행이 되고 Consumer는 메시지를 받아서 DB에 insert만 하는 프로세스라면 메시지간의 영향도가 없으므로 어떤 서버든 Consumer가 받아서 언제 처리가 되든 영향이 없다.

이럴 때는 RoundRobin 방식으로 메시지를 소비하는 게 메시지 처리 퍼포먼스가 좋아진다.

MQ RoundRobin 방식

activemq queue

Queue(개발자가 정의한 destination)에 연결 된 Consumer가 3개라면 Message1은 Consumer1이 수신받고 Messaage2는 Consumer2이 Message3은 Consumer3이 수신받는데 한 번씩 배분되었으면 다시 Message4는 Consumer1이 수신받는다.

메시지들을 Consumer에게 동일하게 배분하여 순환 처리하는 방식이다.

 

Consumer 서버가 클러스터링 환경일 경우 Consumer(Thread)가 서버 한대당 한 개 일 수도 있고 다수 일 수도 있다. 메시지양과 메시지 처리율에 따른 서버 리소스 상태 등을 고려하여 설정해야 한다.

Consumer Method에서 어노테이션 설정에서 concurrency 옵션으로 Consumer(Thread) 수를 정의할 수 있다.

@JmsListener(destination = "브로커 큐 이름", concurrency = "1-2")

concurrenty = "1-2"의 의미는 JmsListener가 시작될 때 Consumer(Thread) 수를 최소 1개에서 최대 2개까지 생성하겠다는 정의이다. 이렇게 하면 메시지 처리 병목 상태에 따라 컨슈머 수를 제어하게 된다. 물론 concurrency = "2" 이런 식으로 최댓값이 변하지 않게 고정하는 방식을 정의할 수도 있다.

concurrency = "2"로 설정되고 클러스터링에 의한 서버 수가 2대라면 총 4개의  Consumer를 운영하게 된다.

Message B : 메시지간의 영향이 있는 메시지

예를 들어 상품의 재고 값을 업데이트해야 하는 정보가 담긴 메시지가 Producer로부터 발행이 되고 Consumer는 메시지를 받아서 DB에 해당 상품의 재고 값을 업데이트해야 한다면 메시지간의 영향이 발생되므로 Consumer의 처리 순서가 매우 중요해진다.

이때 라운드 로빈 방식으로 메시지를 처리하게 되면 위험도가 올라간다.

각 서버마다 서버의 내부적, 외부적 요인에 따라 메시지 처리 속도가 달라질 수 있기 때문이다.

Message B를 Round Robin으로 처리하면 위험한 이유

activemq consumer

Message	상품ID	재고
Message1	1	7
Message2	1	5
Message3	1	2

Queue에 쌓인 메시지 1,2,3 모두 정상적으로 처리가 되었다면 최종적으로 원하는 재고 값은 2가 되어야 할 것이다.

 

Consumer1 -> Consumer2 -> Consumer3 순서대로 Message1 -> Message2 -> Message3 순서대로 처리되면 문제가 없겠지만 서버마다 Consumer 처리속도는 달라질 수 있다. 그래서 어떤 메시지가 먼저 처리될지는 장담할 수 없다.

위의 그림으로 예를 들면 Message를 Consumer들에게 하나씩 순서대로 전달했지만 메시지 처리는 Message2->Message3->Message1 순서로 처리가 됐을 것이다. 그럼 최종적으로 업데이트된 재고 값은 7이 될 것이다.

 

이때는 메시지 처리 속도보다는 메시지 우선순위 처리에 중점을 둬야 한다.

ActiveMQ에는 MessageGroup이라는 기능을 제공한다. MessageGroup을 사용하게 되면 다수의 Consumer 중 먼저 Connection 된 Consumer와만 연결이 되어 1:1 통신을 하게 된다.

activemq consumer

 

jmsTemplate.convertAndSend(DESTINATION, message, mqMessage -> {
	mqMessage.setStringProperty("JMSXGroupID", "stock"); // stock이라는 그룹아이디를 부여
	return mqMessage;
});

참고사항 : ActiveMq Message Group Document

Message 객체에 JMSXGroupID 속성 값을 설정하면 그룹 아이디로 그룹화되어 한 개의 Consumer에서만 연결되어 메시지를 처리하게 된다. Message1이 처리된 이후에 Message2를 수신받게 되니 우선순위가 명확해진다. 다만 위에도 언급했듯이 RoundRobin 방식보다는 메시지 소비 속도 현저히 떨어지게 된다.

그럼에도 불구하고 Consumer2, Consumer3이 낭비되는 것이 싫었고 모든 메시지를 Consumer1만 수신받게 되어 Consumer1의 서버 부하에 대해 무시할 순 없었다.

MessageGroup ID로 Consumer 분산 처리

재고 메시지 정보를 기준으로 한다면 JMSXGroupID Value에 stock-"상품아이디"를 설정해보자.

jmsTemplate.convertAndSend(DESTINATION, message, mqMessage -> {
	mqMessage.setStringProperty("JMSXGroupID", "stock-" + stock.getProductId()); // 재고 상품아이디를 그룹화
	return mqMessage;
});

 

activemq consumer

그룹화된 상품ID 별로 Consumer에게 메시지를 할당해 주기 때문에 위의 이미지와 같이 Consumer들은 할당된 상품ID 메시지만 받아서 처리가 가능해진다. 이 방식으로 서로 영향이 발생되는 메시지 별로 우선순위를 지키며 처리가 가능하며 모든 메시지들은 Consumer 별로 분산처리가 되니 메시지 처리 속도 효율이 매우 좋아진다.

 

 

ActiveMQ 메시지 처리 방식에 대해 기록했지만 모든 큐 방식이 고려해야하는 사항이기도 하다. MQ를 적용하여 개발하려고 할 때에 메시지의 정보 성향과 메시지 간의 영향도, Consumer 처리에 따른 서버 부하 대해 고려해야 할 사항이다.