안녕하세요. 엔지엠소프트웨어입니다. 소켓 통신을 직접 개발하기는 쉽지 않은 일입니다. 고려해야 할 사항이 너무 많기 때문이죠~ 그래서, ActiveMQ나 TibRV와 같은 미들웨어를 사용합니다. 이 글을 검색하셔서 찾아오신 분들은 아마 개발자 분들일테니 TCP, UDP에 대해 잘 알고 있을겁니다. TCP와 같은 속도에 UDP와 같은 안전성을 보장하기 위해 ActiveMQ, TibRV, RabbitMQ등등... 사용합니다.

 

아래는 ActiveMQ의 Server 코드입니다. 여기서 중요한 부분은 메세지를 수신할 이벤트를 등록해야 한다는겁니다. consumer에 listener를 등록 해줍니다. listener는 OnMessage 이벤트 핸들러를 인자로 받습니다. 그리고, 프로듀서를 벗어나지 못하게 반복기(while)가 추가되어 있습니다.

onnecturi = new Uri($"tcp://{_activeMqHost}:{_activeMqPort}");
IConnectionFactory factory = new NMSConnectionFactory(connecturi);

using (IConnection connection = factory.CreateConnection())
{
    using (ISession session = connection.CreateSession())
    {
        IDestination destination = SessionUtil.GetDestination(session, $"queue://{_activeMqQueueName}");

        using (IMessageConsumer consumer = session.CreateConsumer(destination))
        {
            using (IMessageProducer producer = session.CreateProducer(destination))
            {
                var task = producer.SendAsync(null);

                connection.Start();
                producer.DeliveryMode = MsgDeliveryMode.Persistent;
                producer.RequestTimeout = _receiveTimeout;

                consumer.Listener += new MessageListener(OnMessage);

                while (true)
                    System.Threading.Thread.Sleep(100);
            }
        }
    }
}

 

OnMessage 이벤트 핸들러에서 메세지를 파싱하고, 비즈니스 로직들을 처리 해줍니다.

protected void OnMessage(IMessage receivedMsg)
{
    var msg = receivedMsg as ITextMessage;
    if (msg != null)
    {
        WriteActiveMqLog(msg.Text);
        DCP_ActiveMqMessage(msg.Text, receivedMsg);
    }
}

 

위의 비즈니스 로직을 처리하는 메소드는 DCP_ActiveMqMessage입니다. 클라이언트로부터 받은 메세지를 처리하고, ReplyTo로 전달하기 위한 코드는 아래와 같습니다.

var model = JsonConvert.DeserializeObject<Common.Models.RealTimeModel>(msg);
var realTimeModel = RealTimeParameterMapping(model);

Uri connecturi = new Uri($"tcp://{_activeMqHost}:{_activeMqPort}");
IConnectionFactory factory = new NMSConnectionFactory(connecturi);

using (IConnection connection = factory.CreateConnection())
{
    using (ISession session = connection.CreateSession())
    {
        using (IMessageProducer producer = session.CreateProducer(iMessage.NMSReplyTo))
        {
            ITextMessage reponse = session.CreateTextMessage(JsonConvert.SerializeObject(realTimeModel));
            reponse.NMSCorrelationID = iMessage.NMSCorrelationID;

            producer.Send(reponse);
        }
    }
}

 

ActiveMQ를 실행한 후 크롬에서 localhost:8161로 접속 해보세요. 관리자 아이디와 패스워드는 admin/admin입니다. 아래 그림과 같은 화면에서 Manager ActiveMQ Broker를 클릭하세요.

 

 

Queues를 클릭하세요.

 

 

서버와 클라이언트가 메세지를 주고 받는 test1 큐를 하나 생성 했습니다. 해당 큐의 Send To를 클릭하세요.

 

 

아래 그림과 같이 서버로 보낼 메세지를 입력하고, Send 버튼을 클릭하세요.

 

 

이렇게하면 서버로 메세지는 전달되지만~ 서버에서 처리된 결과 메세지를 받을수는 없습니다. TibRV의 SendWait와 같이 동작하려면 ActiveMQ는 ReplyTo를 사용해야 합니다. 아래와 같이 Correlation ID에 동기화에 사용할 유니크한 아이디를 입력하세요. 프로그래밍에서는 보통 yyyyMMddHHmmssfff와 같이 중복되지 않게 사용합니다. 또는 GUID를 생성해서 넣어줘도 됩니다.

 

 

다시 Send 버튼을 클릭하세요.

 

 

클라이언트에서 ReplyTo에 temp1큐를 설정 했습니다. 서버에서 이 값을 그대로 데스티네이션에 넣고 Send로 보내줍니다.

using (IMessageProducer producer = session.CreateProducer(iMessage.NMSReplyTo))
{
    ITextMessage reponse = session.CreateTextMessage(JsonConvert.SerializeObject(realTimeModel));
    reponse.NMSCorrelationID = iMessage.NMSCorrelationID;

    producer.Send(reponse);
}

 

temp1 큐를 보면 서버로부터 메세지를 받았다는걸 알 수 있습니다.

 

 

클라이언트에서도 프로듀서와 콘슈머를 둘다 만들어야 합니다. 그리고, temp1 큐에 들어온 메세지를 빼와서 추가적인 비즈니스 로직을 만들어야 하겠죠? 이렇게하면 서버와 클라이언트가 메세지를 주고 받을 수 있고, 코릴레이션 아이디로 메세지를 동기화할 수 있습니다. 좀 더 복잡하게 처리할 수 있는 방법도 있지만, 이게 가장 이상적이고 간단한 방법인듯 합니다.

 

개발자에게 후원하기

 

추천, 구독, 홍보 꼭~ 부탁드립니다.

여러분의 후원이 빠른 귀농을 가능하게 해줍니다~ 답답한 도시를 벗어나 귀농하고 싶은 개발자~

감사합니다~

추천0 비추천0