[2022.02] MSMQ를 이용해 프로세스 간 메시지 송/수신 처리하기

안녕하세요! 오랜만에 글을 작성해봅니다. 올해 처음인가봐요 ㅡㅡ; (다행히 아닙니다. ㅎㅎ)

 

MSMQ는 윈도우 OS에서 제공하는 Microsoft Message Queue 입니다. 이를 사용하기 위해서는 제어판 > 윈도우 기능 켜기/끄기에서 활성화 해주셔야 합니다. 

본 글에서 테스트 한 메시지 형태는 클래스 형태로 만들어 상호 교환되는 것을 확인하는 소스입니다. 

XML방식을 직렬화 및 역직렬화를 하는데 속도가 느리다는 글이 있네요. 그래서 아래 참조 글 중 JSON방식의 포맷터를 사용한 테스트 코드를 올려봅니다. 

 

다음 두 개의 클래스를 추가합니다. 

public class Order
{
    public long orderId;
    public DateTime orderTime;
};

 

public class EnDevMessageQueue
    {
        /// <summary>
        /// MSMQ에 QueuePath 주소가 있는지 확인하고 없으면 생성
        /// </summary>
        /// <param name="QueuePath">ex: @".\Private$\EndevTest"</param>
        public static void CreateQueue(string QueuePath)
        {
            if (MessageQueue.Exists(QueuePath))
            {
                MessageQueue.Delete(QueuePath);
                MessageQueue.Create(QueuePath);
            }
        }

        public static void SendQueue(string QueuePath, string Label, object SendMsg)
        {
            MessageQueue messageQueue = null;

            if (MessageQueue.Exists(QueuePath))
            {
                messageQueue = new MessageQueue(QueuePath);
                messageQueue.Label = Label;
            }
            else
            {
                // Create the Queue (큐생성없이 SendQueue를 실행했을 때의 예외처리)
                MessageQueue.Create(QueuePath);
                messageQueue = new MessageQueue(QueuePath);
                messageQueue.Label = Label;
            }
            messageQueue.Formatter = new JsonMessageFormatter();
            messageQueue.Send(SendMsg);
        }

        public static object[] RecvQueue(string QueuePath)
        {
            MessageQueue messageQueue = new MessageQueue(QueuePath);
            messageQueue.Formatter = new JsonMessageFormatter();

            Message[] messages = messageQueue.GetAllMessages();
            List<object> lstObj = new List<object>();

            foreach (Message message in messages)
            {
                lstObj.Add(message.Body);
            }
            
            return lstObj.ToArray();
        }

        public static Message[] GetMessages(string QueuePath)
        {
            MessageQueue messageQueue = new MessageQueue(QueuePath);
            messageQueue.Formatter = new JsonMessageFormatter();
            return messageQueue.GetAllMessages(); 
        }

        public static int Count(string QueuePath)
        {
            MessageQueue messageQueue = new MessageQueue(QueuePath);
            return messageQueue.GetAllMessages().Length;
        }

        public static void PurgeQueue(string QueuePath)
        {
            MessageQueue messageQueue = new MessageQueue(QueuePath);
            // after all processing, delete all the messages
            messageQueue.Purge();
        }
    }

사용방법 (메시지 추가)

Order order = new Order();
order.orderId = DateTime.Now.ToFileTimeUtc;
order.orderTime = DateTime.Now;

EndevMessageQueue.SendQueue(@"\Private$\EndevTest", "EndevTest", order);

 

사용방법 (메시지 처리)

// 메시지 큐에 있는 모든 메시지 가져오기
Object[] orders = EnDevMessageQueue.RecvQueue(@".\Private$\EndevTest");
foreach (Object ord in orders)
{
    Order order = cfg as Order;
    Console.WriteLine(order.orderTime.ToString("yyyy-MM-dd HH:mm:ss.fff"));
}

// 전체 메시지 삭제
EnDevMessageQueue.PurgeQueue(@".\Private$\EndevTest");

 

이제 숙제는 메시지 ID별로 처리하고 제거하는 형태로 만들어봐야겠네요. Rabbit MQ사용하다가 MSMQ를 사용하니 편의성이 현저하게 떨어지는... 다만 OS에서 지원하기 때문에 프로그램이 종료되어도 남아있는 장점이 있습니다. RabbitMQ의 경우 메시지 큐 프로세스가 종료되면 직전까지 가지고 있던 것을 파일로 가지고 있어서 재시작하면 종료되기 직전까지 가지고 있던 메시지를 처리할 수 있거든요. 

 

참고) 

https://docs.microsoft.com/ko-kr/dotnet/api/system.messaging.messagequeue.formatter?view=netframework-4.7.2 

MessageQueue.Formatter 속성 (System.Messaging)

 

https://gist.github.com/jchadwick/2430984

MSMQ Message JSON Formatter