이전 글에서는 Azure Messaging의 중복 메시지 처리 메커니즘인 '최대 한 번(at-most-once)'과 '최소 한 번(at-least-once)'에 대해 다뤘습니다.

Azure Messaging-ServiceBus Messaging 메시지 큐 기술 시리즈 5-중복 메시지: at-least-once at-most-once

이번 글에서는 Azure Messaging의 메시지 확인(Receipt) 메커니즘을 살펴보겠습니다. 실제 적용 사례는 다음과 같습니다:

동기 수신/송신 시나리오에서 메시지 발신자와 수신자 간의 양방향 응답 패턴입니다. 예를 들어, 장삼이 편지를 우체국 중계소에 보내면, 이사가 중계소에서 편지를 받고 회신을 작성하여 다시 중계소에 보냅니다. 물론 장삼이 편지를 쓸 때 회신 주소를 명시해야 합니다. 또 다른 예로, 주문 번호를 생성하는 시나리오가 있습니다. 주문 번호를 생성하는 메시지를 발송하고, 메시지 수신자가 이를 처리한 후 메시지 확인을 통해 결과를 반환합니다.

Azure Messaging의 메시지 확인 메커니즘은 주로 세션(Session)을 지원하는 Queue/Topic, SessionId, ReplyTo 속성을 기반으로 구현됩니다.

코드 구현을 위해 다음 사항이 필요합니다:

1. 두 개의 작업 스레드: 하나는 메시지 발송 및 확인 메시지 수신용, 다른 하나는 메시지 수신 및 확인 메시지 발송용입니다.
2. 하나의 세션 식별자: ReceiptSession
3. 두 개의 큐(Queue): RequestQueue(메시지 발송/수신용), ResponseQueue(확인 메시지 발송/수신용)

코드 예제를 바로 살펴보겠습니다:

먼저, ServiceBusHelper 클래스에 스레드로부터 안전하게 세션을 지원하는 QueueClient를 생성하는 메서드를 추가합니다:

private static object queueCreationLock = new object();
/// <summary>
/// 세션을 요구하는 QueueClient를 반환합니다.
/// </summary>
/// <param name="queueName">큐 이름</param>
/// <returns>QueueClient</returns>
public QueueClient GetSessionEnabledQueueClient(string queueName)
{
    var namespaceClient = NamespaceManager.Create();
    if (!namespaceClient.QueueExists(queueName))
    {
        lock (queueCreationLock)
        {
            if (!namespaceClient.QueueExists(queueName))
            {
                var queue = new QueueDescription(queueName) { RequiresSession = true };
                namespaceClient.CreateQueue(queue);
            }
        }
    }

    return QueueClient.Create(queueName, ReceiveMode.ReceiveAndDelete);
}

다음으로 몇 가지 상수를 정의합니다:

private static readonly string receiptSessionId = "ReceiptSession";

const double responseMessageTimeout = 20.0;

private static readonly string requestQueue = "RequestQueue";

private static readonly string responseQueue = "ResponseQueue";

메시지를 발송하고 확인 메시지를 수신하는 메서드를 구현합니다:

/// <summary>
/// 메시지를 발송하고 확인 메시지를 수신합니다.
/// </summary>
public static void SendRequestMessage()
{
    var helper = new ServiceBusHelper();
    var responseQueueClient = helper.GetSessionEnabledQueueClient(responseQueue);
    var requestQueueClient = helper.GetSessionEnabledQueueClient(requestQueue);

    var receiptReceiver = responseQueueClient.AcceptMessageSession(receiptSessionId);
    var order = CreateOrder(1); // SalesOrder 대신 Order 클래스 사용

    // 메시지 발송
    var message = new BrokeredMessage(order);
    message.Properties.Add("Type", order.GetType().ToString());
    message.SessionId = receiptSessionId;
    message.MessageId = "OrderMessage001";
    message.ReplyTo = responseQueue;
    requestQueueClient.Send(message);
    Console.WriteLine("메시지 발송: " + message.MessageId + ", 주문 ID: " + order.OrderID);

    // 확인 메시지 수신
    var receivedReceipt = receiptReceiver.Receive(TimeSpan.FromSeconds(responseMessageTimeout * 2));

    var receivedOrder = receivedReceipt.GetBody<Order>();
    Console.WriteLine("확인 메시지 수신: " + receivedReceipt.MessageId + ", 주문 ID: " + receivedOrder.OrderID);
    receiptReceiver.Close();
}

메시지를 수신하고 확인 메시지를 발송하는 메서드를 구현합니다:

/// <summary>
/// 메시지를 수신하고 확인 메시지를 발송합니다.
/// </summary>
public static void ProcessRequestAndSendReceipt()
{
    var helper = new ServiceBusHelper();

    var requestQueueClient = helper.GetSessionEnabledQueueClient(requestQueue);
    var requestSession = requestQueueClient.AcceptMessageSession();
    var incomingMessage = requestSession.Receive();

    if (incomingMessage != null)
    {
        var receivedOrder = incomingMessage.GetBody<Order>();
        Console.WriteLine("메시지 수신: " + incomingMessage.MessageId + ", 주문 ID: " + receivedOrder.OrderID);

        var outgoingReceipt = new BrokeredMessage(receivedOrder);
        outgoingReceipt.Properties.Add("Type", receivedOrder.GetType().ToString());
        outgoingReceipt.ReplyToSessionId = receiptSessionId;
        outgoingReceipt.MessageId = "ResponseOrderMessage001";
        outgoingReceipt.SessionId = incomingMessage.SessionId;
        
        // 확인 메시지 발송
        var responseQueueClient = helper.GetSessionEnabledQueueClient(incomingMessage.ReplyTo);
        responseQueueClient.Send(outgoingReceipt);
        Console.WriteLine("확인 메시지 발송: " + outgoingReceipt.MessageId + ", 주문 ID: " + receivedOrder.OrderID);               
    }
}

Main 메서드에서 두 개의 작업 스레드를 시작합니다: 하나는 메시지 발송 및 확인 메시지 수신용, 다른 하나는 메시지 수신 및 확인 메시지 발송용입니다.

Azure Messaging에서 큐는 처음 생성 시 여러 요청이 동시에 동일한 큐를 생성하는 것을 지원하지 않으므로, 두 스레드 간에 간단한 Task.Delay(3000).Wait()를 사용하여 시간차를 두었습니다.

static void Main(string[] args)
{
    var sendTask = Task.Run(() => { SendRequestMessage(); });
    Task.Delay(3000).Wait();
    var receiveTask = Task.Run(() => { ProcessRequestAndSendReceipt(); });

    Task.WaitAll(sendTask, receiveTask);

    Console.ReadKey();           
}

프로그램 출력을 확인해 보면:

Azure 서비스 버스의 큐:

위 결과에서 알 수 있듯이, Azure Messaging-ServiceBus Messaging은 세션을 지원하는 Queue/Topic, SessionId, ReplyTo 속성을 기반으로 메시지 확인 메커니즘을 구현합니다.