Message Queue (MQ)

메시키 큐란?

데이터를 처리하는 과정이 크게 두 가지로 나뉜다. 실시간으로 처리하는 온라인 작업, 그리고 일괄적으로 모아서 처리하는 배치 작업. 메시지 큐는 배치 작업을 위해 유용하게 사용된다.

메시지 큐는 기본적으로 메시지를 전달하기 위한 임시 저장소이며, 다음과 같은 구성 요소로 이루어져 있다.

1. 생산자 (Producer): 메시지를 생성하고 큐에 보내는 역할
2. 메시키 큐 (Message Queue): 생산자로부터 메시지를 받아 저장하고 있다가, 소비자가 준비되면 메시지를 전달
3. 소비자 (Consumer): 큐에서 메시지를 받아 처리

 

 

이 구조에서 중간에 메시지 큐가 끼어있음으로 생기는 이점은

• 비동기: 생산자와 소비자가 메시지를 동시에 처리하지 않아도 됨. 온라인 작업과는 다르게 배치 작업은 대개의 경우 양쪽 프로세스가 깨어있지 않은 상태에서 (특히 밤에) 돌아가기 때문에 비동기 통신이 중요함.
• 탄력성: 시스템 장애가 발생해도 메시지가 손실되지 않는다. 메시지는 메시지 큐에 남아 있기 때문에.
• 낮은 결합도: 생산자와 소비자가 독립적으로 행동하게 됨으로써 서비스들 간 결합도가 낮아짐.

 

---

구현 예시 (IBM MQ)

아래 코드는 IBM MQ에 연결하고 큐를 관리하는 Java 클래스를 구현한 예시. IBM MQ 외에도 RabbitMQ, ActiveMQ, Apache Kafka 등 여러 유사한 메시지 큐잉 시스템이 있다. 

 

IbmMqImpl.java

/*
 * IBM MQ general methods including:
 * - client connection
 * - browse, get, put messages
 * - commit, rollback
 * - release
 */

import java.io.*;
import java.security.*;
import java.util.*;

import com.ibm.mq.*;
import com.ibm.mq.constants.CMQC;

public class IbmMqImpl {
	protected static int retryInterval = 60 * 1000;
	protected static int retryMax = 3;

	private MQQueueManager queueManager;
	private String queueManagerName;
	private String ccdt; // client channel definition table

	private Map<String, MQQueue> putQueues = new Hashtable<String, MQQueue>();
	private Map<String, MQQueue> getQueues = new Hashtable<String, MQQueue>();
	private Map<String, MQQueue> browseQueues = new Hashtable<String, MQQueue>();

	// Creates a connection to the named queue manager using a channel definition table.
	public IbmMQImpl(String queueManagerName, String ccdt) throws IbmMqException {
		this.queueManagerName = queueManagerName;
		this.ccdt = ccdt;
		java.net.URL url = IbmMqImpl.class.getClassLoader().getResource(ccdt);
		
		try {
			url.openConnection();
		} catch (Exception e) {
			e.printStackTrace();
			throw new IbmMqException("An error occured while connecting " + ccdt + ".", e);
		}
		
		boolean finish = false;
		int retryCnt = 0;
		
		while(!finish) {
			try {
				queueManager = new MQQueueManager(queueManagerName, url);
				finish = true;
			} catch (MQException e) {
				if (e.reasonCode == CMQC.MQRC_CONNECTION_BROKEN) {
					try {
						retryCnt++;
						if (retryCnt >= retryMax) {
							finish = true;
						}
						Thread.sleep(retryInterval);
					} catch (InterruptedException e1) {
						throw new IbmMqException("An error occured while connecting to " + queueManagerName + ": " + e.reasonCode, e);
					}
				} else {
					throw new IbmMqException("An error occured while connecting to " + queueManagerName + ": " + e.reasonCode, e);
				}
			}
		}
	}

	private MQQueue getBrowseQueue(String queueName) throws MQException {
		MQQueue queue = null;
		queue = (MQQueue) browseQueues.get(queueName);
		if (queue == null) {
			int openOptions = CMQC.MQOO_BROWSE | CMQC.MQOO_INPUT_SHARED | CMQC.MQOO_FAIL_IF_QUIESCING
					| CMQC.MQOO_BIND_NOT_FIXED;
			queue = queueManager.accessQueue(queueName, openOptions);
			browseQueues.put(queueName, queue);
		}
		return queue;
	}

	private MQQueue getGetQueue(String queueName) throws MQException {
		MQQueue queue = null;
		queue = (MQQueue) getQueues.get(queueName);
		if (queue == null) {
			int openOptions = CMQC.MQOO_INPUT_AD_Q_DEF | CMQC.MQOO_INPUT_SHARED | CMQC.MQOO_FAIL_IF_QUIESCING
					| CMQC.MQOO_BIND_NOT_FIXED;
			queue = queueManager.accessQueue(queueName, openOptions);
			getQueues.put(queueName, queue);
		}
		return queue;
	}

	private MQQueue getPutQueue(String queueName) throws MQException {
		MQQueue queue = null;
		queue = (MQQueue) putQueues.get(queueName);
		if (queue == null) {
			int openOptions = CMQC.MQOO_OUTPUT | CMQC.MQOO_FAIL_IF_QUIESCING | CMQC.MQOO_BIND_ON_OPEN;
			queue = queueManager.accessQueue(queueName, openOptions);
			putQueues.put(queueName, queue);
		}
		return queue;
	}

	public byte[] browseMessage(String queueName) throws IbmMqException {
		byte[] msg = null;
		MQQueue browseQueue = null;

		try {
			browseQueue = getBrowseQueue(queueName);
			MQGetMessageOptions gmo = new MQGetMessageOptions();
			gmo.options += CMQC.MQGMO_BROWSE_FIRST;
			gmo.options += CMQC.MQGMO_WAIT;
			gmo.options += CMQC.MQGMO_FAIL_IF_QUIESCING;
			gmo.options += CMQC.MQGMO_COMPLETE_MSG;
			gmo.waitInterval = 3000;

			MQMessage inMsg = new MQMessage();

			browseQueue.get(inMsg, gmo);

			msg = new byte[inMsg.getDataLength()];
			inMsg.readFully(msg);
		} catch (MQException e) {
			throw new IbmMqException("MQException occured while browsing the message.", e);
		} catch (IOEXception e) {
			throw new IbmMqException("IOEXception occured while browsing the message.", e);
		} catch (NumberFormatException e) {
			throw new IbmMqException("NumberFormatException occured while browsing the message.", e);
		}
		return msg;
	}

	public void getMessage(String queueName) throws IbmMqException {
		byte[] msg = null;
		MQQueue getQueue = null;

		try {
			getQueue = getGetQueue(queueName);
			MQGetMessageOptions gmo = new MQGetMessageOptions();
			gmo.options += CMQC.MQGMO_BROWSE_FIRST;
			gmo.options += CMQC.MQGMO_WAIT;
			gmo.options += CMQC.MQGMO_FAIL_IF_QUIESCING;
			gmo.options += CMQC.MQGMO_COMPLETE_MSG;
			gmo.waitInterval = 3000;

			MQMessage inMsg = new MQMessage();

			getQueue.get(inMsg, gmo);

			msg = new byte[inMsg.getDataLength()];
			inMsg.readFully(msg);
		} catch (MQException e) {
			throw new IbmMqException("MQException occured while getting the message.", e);
		} catch (IOEXception e) {
			throw new IbmMqException("IOEXception occured while getting the message.", e);
		}
		return msg;
	}

	public void putMessage(byte[] msg, String queueName) throws IbmMqException {
		MQQueue putQueue = null;
		try {
			putQueue = getPutQueue(queueName);
			MQPutMessageOptions mpmo = new MQPutMessageOptions();
			mpmo.options += CMQC.MQMO_FAIL_IF_QUIESCING;

			MQMessage mqMessage = nQMessage();
			mqMessage.messageFlags = CMQC.MQMF_SEew MGMENTATION_ALLOWED;

			mqMessage.characterSet = 1208;
			mqMessage.write(msg);

			putQueue.put(mqMessage, mpmo);
		} catch (MQException e) {
			throw new IbmMqException("MQException occured while sending the message.", e);
		} catch (IOEXception e) {
			throw new IbmMqException("IOEXception occured while sending the message.", e);
		}
	}

	public void deleteMessage(String queueName) throws IbmMqException {
		MQQueue getQueue = null;
		try {
			getQueue = getGetQueue(queueName);
			
			MQGetMessageOptions gmo - new MQGetMessageOptions();
			gmo.options += CMQC.MQGMO_FAIL_IF_QUIESCING;
			gmo.options += CMQC.MQGMO_COMPLETE_MSG;
			
			MQMessage inMsg = new MQMessage();
			getQueue.get(inMsg, gmo);
		} catch (MQException e) {
			throw new IbmMqException("MQException occured while deleting the message.", e);
		}
	}

	public void commit() throws IbmMqException {
		if (queueManager != null) {
			try {
				queueManager.commit();
			} catch (MQException e) {
				throw new IbmMqException("MQException occured while committing the resource.", e);
			}
		}
	}

	public void rollback() throws IbmMqException {
		if (queueManager != null) {
			try {
				queueManager.backout();
			} catch (MQException e) {
				throw new IbmMqException("MQException occured while rollback the resource.", e);
			}
		}
	}

	private void releaseQueue(Map<String, MQQueue> queues) throws IbmMqException {
		for (Iterator<String> it = queues.ketSet().iterator(); it.hasNext();) {
			String queueName = (String) it.next();
			MQQueue mqQueue = queues.get(queueName);

			if (mqQueue != null) {
				try {
					mqQueue.close();
				} catch (MQException e) {
					mqQueue = null;
					throw new IbmMqException("MQException occured while releasing the resource.", e);
				}
				mqQueue = null;
			}
		}
		queues.clear();
	}

	public void release() throws IbmMqException {
		releaseQueue(putQueues);
		releaseQueue(getQueues);
		releaseQueue(browseQueues);

		if (queueManager != null) {
			try {
				queueManager.close();
				queueManager.disconnect();
			} catch (MQException e) {
				queueManager = null;
				throw new IbmMqException("MQException occured while releasing the resource.", e);
			}
			queueManager = null;
		}
	}
}

 

IbmMqException.java

public class IbmMqException extends Exception {
	public IbmMqException(String msg, Exception e) {
		super(msg);
	}
}