Microsoft MVP성태의 닷넷 이야기
글쓴 사람
정성태 (techsharer at outlook.com)
홈페이지
첨부 파일

C# - MQTT를 이용한 클라이언트/서버(Broker) 통신 예제

예전에 소개한 CoAP처럼,

C# - CoAP 서버 및 클라이언트 제작 (UDP 소켓 통신)
; https://www.sysnet.pe.kr/2/0/12629

MQTT 또한 사물 인터넷 용으로 나온 프로토콜이긴 하지만 여타 응용 프로그램에서 사용하지 않을 이유가 없습니다. ^^ 그리고 CoAP와 다른 점이 있는데요. 일반적인 서버/클라이언트 개념으로 보면, 서버 역할을 MQTT broker가, 클라이언트 역할을 MQTT client가 수행합니다. 하지만, 서버/클라이언트처럼 서버 측에서 서비스를 제공하는 방식이 아니고, broker와 연결된 특정 클라이언트가 서비스를 수행하는 주체가 될 수 있습니다. 즉, MQTT broker는 말 그대로 통신 간의 중계 역할에 더 초점을 둡니다.

비슷한 예로, WPF의 Prism 프레임워크에서 사용하는 Event Subscriber/Publisher의 개념과 유사하다고 보면 되겠습니다.

실제 코드로 간단한 예를 들면 더 편하겠지요. ^^ .NET에서 MQTT 프로토콜을 구현한 MQTTnet 라이브러리를 사용하면 더 쉽게 접근할 수 있으니,

chkr1011/MQTTnet
; https://github.com/chkr1011/MQTTnet

Getting Started with MQTT
; https://www.hivemq.com/blog/how-to-get-started-with-mqtt/

Example of MQTT communication using MQTTnet
; https://www.programmersought.com/article/65964914395/

landbroken/MQTTLearning
; https://github.com/landbroken/MQTTLearning

이것을 이용해 예제를 만들어 보겠습니다. 그런데, 의외로 웹 상에 공개된 예제들이 2.x 버전의 것이라 현재 3.x 버전으로는 빌드가 안 됩니다. 하위 호환성이 ^^; 꽝인데 업그레이드 가이드에서 기록을 찾아볼 수 있습니다.

이를 기반으로 우선 서버 먼저 다음과 같이 간단하게 만들 수 있습니다.

// Install-Package MQTTnet

using MQTTnet;
using MQTTnet.Client.Receiving;
using MQTTnet.Server;
using System;
using System.Text;
using System.Threading.Tasks;

namespace MqttSample
{
    class Program
    {
        static async Task Main(string[] args)
        {
            var optionsBuilder = new MqttServerOptionsBuilder()
                    .WithConnectionBacklog(100)
                    .WithDefaultEndpointPort(8222);

            IMqttServer broker = new MqttFactory().CreateMqttServer();

            broker.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e => OnMessageReceived(e));
            broker.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(e => OnClientConnected(e));
            broker.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(e => OnClientDisconnected(e));
            broker.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(e => OnClientSubscribedToTopic(e));
            broker.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(e => OnClientUnsubscribedToTopic(e));

            await broker.StartAsync(optionsBuilder.Build());

            Console.WriteLine("Press any key to exit...");
            Console.ReadLine();
        }

        private static void OnClientUnsubscribedToTopic(MqttServerClientUnsubscribedTopicEventArgs e)
        {
            Console.WriteLine($"OnClientUnsubscribedToTopic: {e.ClientId} - {e.TopicFilter}");
        }

        private static void OnClientSubscribedToTopic(MqttServerClientSubscribedTopicEventArgs e)
        {
            Console.WriteLine($"OnClientSubscribedToTopic: {e.ClientId} - {e.TopicFilter.Topic}");
        }

        private static void OnMessageReceived(MqttApplicationMessageReceivedEventArgs e)
        {
            Console.WriteLine($"OnMessageReceived: {e.ApplicationMessage.Topic}, {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
        }

        private static void OnClientConnected(MqttServerClientConnectedEventArgs e)
        {
            Console.WriteLine($"OnClientConnected: {e.ClientId}");
        }

        private static void OnClientDisconnected(MqttServerClientDisconnectedEventArgs e)
        {
            Console.WriteLine($"OnClientDisconnected: {e.ClientId} - {e.DisconnectType}");
        }
    }
}

그런데, 디버깅 용도의 로그 정보 출력에 해당하는 이벤트 구독은 안 해도 됩니다. 그래서 서버는 다음과 같이 간단하게 만들어도 무방합니다.

using MQTTnet;
using MQTTnet.Server;
using System;
using System.Threading.Tasks;

namespace MqttSample
{
    class Program
    {
        static async Task Main(string[] args)
        {
            var optionsBuilder = new MqttServerOptionsBuilder()
                    .WithConnectionBacklog(100)
                    .WithDefaultEndpointPort(8222);

            IMqttServer broker = new MqttFactory().CreateMqttServer();

            await broker.StartAsync(optionsBuilder.Build());

            Console.WriteLine("Press any key to exit...");
            Console.ReadLine();
        }
    }
}

그다음, 위의 Broker에 Topic을 발행하고/구독할 클라이언트를 만들어야 하는데요. 이것은 MQTTnet을 기반으로 클라이언트를 좀 더 만들기 쉽게 래퍼한 MQTTnet.Extensions.ManagedClient 패키지를 이용하면 더 쉽게 만들 수 있습니다. 따라서 다음의 소스 코드를 기반으로,

MQTTnet/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs /
; https://github.com/chkr1011/MQTTnet/blob/master/Tests/MQTTnet.TestApp.NetCore/ManagedClientTest.cs

이렇게 만들 수 있습니다.

// Install-Package MQTTnet
// Install-Package MQTTnet.Extensions.ManagedClient

using MQTTnet;
using MQTTnet.Client.Options;
using MQTTnet.Client.Receiving;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Protocol;
using System;
using System.Threading.Tasks;

namespace MqttClient
{
    class Program
    {
        static async Task Main(string[] args)
        {
            ManagedMqttClientOptions options = new ManagedMqttClientOptions();

            options.ClientOptions = new MqttClientOptions()
            {
                ClientId = "MyMqttClient",
                ChannelOptions = new MqttClientTcpOptions
                {
                    Server = "localhost",
                    Port = 8222,
                }
            };

            options.AutoReconnectDelay = TimeSpan.FromSeconds(1);

            try
            {
                var managedClient = new MqttFactory().CreateManagedMqttClient();
                managedClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(e =>
                {
                    Console.WriteLine(">> RECEIVED: " + e.ApplicationMessage.Topic);
                });

                await managedClient.StartAsync(options);

                await managedClient.PublishAsync(builder => builder.WithTopic("Step").WithPayload("1"));
                await managedClient.PublishAsync(builder => builder.WithTopic("Step").WithPayload("2").WithAtLeastOnceQoS());

                await managedClient.SubscribeAsync(new MqttTopicFilter { Topic = "xyz", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });
                await managedClient.SubscribeAsync(new MqttTopicFilter { Topic = "abc", QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce });

                await managedClient.PublishAsync(builder => builder.WithTopic("Step").WithPayload("3"));

                Console.WriteLine("Managed client started.");
                Console.ReadLine();
                await managedClient.UnsubscribeAsync("xyz", "abc");
                await managedClient.StopAsync();
            }
            catch (Exception e)
            {
                Console.WriteLine(e);
            }
        }
    }
}

위의 클라이언트는 "Step" 토픽을 만들어 각각 "1", "2", "3"에 해당하는 문자열을 발행하고 있습니다. 또한 토픽 "xyz", "abc"를 구독하고 있는데요, 만약 구독 토픽에 해당하는 메시지가 수신되면 managedClient.ApplicationMessageReceivedHandler에 등록된 이벤트 핸들러의 코드로 인해 화면에는 "RECEIVED: ..." 문자열이 출력될 것입니다.

일단, 위의 상태로 Broker와 Client를 실행시키면 Broker 측의 콘솔 화면에 다음과 같은 출력을 확인할 수 있습니다.

Press any key to exit...
OnClientConnected: MyMqttClient
OnMessageReceived: Step, 1
OnClientSubscribedToTopic: MyMqttClient - xyz
OnClientSubscribedToTopic: MyMqttClient - abc
OnMessageReceived: Step, 2
OnMessageReceived: Step, 3

// [MqttClient를 종료하면]

OnClientUnsubscribedToTopic: MyMqttClient - xyz
OnClientUnsubscribedToTopic: MyMqttClient - abc
OnClientDisconnected: MyMqttClient - Clean

이처럼, broker를 통해 이벤트를 브로드캐스팅할 수 있는 구조를 제공하는데 WPF의 Prism이 이벤트를 exe 프로세스 내에서 중계하는 구조라면 MQTT는 그 범위가 네트워크로 확장되었다고 보면 됩니다. 어쩌면 살짝 UDP 브로드캐스팅과도 닮았는데요, 물론 기존의 UDP 브로드캐스팅은 연결되지 않았어도 서브넷 범위에만 있다면 통신을 할 수 있는 반면 MQTT는 일단 접속해야 한다는 차이점은 있습니다.

(첨부 파일은 이 글의 예제 코드를 포함합니다.)




실습하면서 짐작할 수 있겠지만, MQTT broker는 그 자체로 특정 서비스에 종속적이지 않게 동작할 수 있습니다. 이런 특성 때문에 검색해 보면 MQTT broker를 클라우드 규모로 확장해 서비스하는 제품들이 있습니다. 그중 하나가 HiveMQ인데요,

Getting Started with MQTT
; https://www.hivemq.com/blog/how-to-get-started-with-mqtt/

HiveMQ 사이트에 보면 MQTT에 대해 자세하게 설명하는 문서들을 찾아볼 수 있습니다. 시간이 되면 저도 ^^ 천천히 한 번 읽어봐야겠습니다.

Everything you need to know about the latest version of MQTT: MQTT 5
; https://www.hivemq.com/mqtt-5/

MQTT Basics
; https://www.hivemq.com/mqtt-essentials/

    Introducing the MQTT Protocol - MQTT Essentials: Part 1
    ; https://www.hivemq.com/blog/mqtt-essentials-part-1-introducing-mqtt/

    Publish & Subscribe - MQTT Essentials: Part 2
    ; https://www.hivemq.com/blog/mqtt-essentials-part2-publish-subscribe/

    MQTT Client and Broker and MQTT Server and Connection Establishment Explained - MQTT Essentials: Part 3
    ; https://www.hivemq.com/blog/mqtt-essentials-part-3-client-broker-connection-establishment/

    MQTT Publish, Subscribe & Unsubscribe - MQTT Essentials: Part 4
    ; https://www.hivemq.com/blog/mqtt-essentials-part-4-mqtt-publish-subscribe-unsubscribe/

    MQTT Topics & Best Practices - MQTT Essentials: Part 5
    ; https://www.hivemq.com/blog/mqtt-essentials-part-5-mqtt-topics-best-practices/
    ; https://medium.com/@jspark141515/mqtt-%EC%8B%A4%EC%A0%84-%EC%9D%91%EC%9A%A9%ED%95%98%EA%B8%B0-wildcard-9643877262f2

    Quality of Service 0,1 & 2 - MQTT Essentials: Part 6
    ; https://www.hivemq.com/blog/mqtt-essentials-part-6-mqtt-quality-of-service-levels/
    ; https://medium.com/@jspark141515/mqtt%EB%9E%80-314472c246ee

    Persistent Session and Queuing Messages - MQTT Essentials: Part 7
    ; https://www.hivemq.com/blog/mqtt-essentials-part-7-persistent-session-queuing-messages/

    Retained Messages - MQTT Essentials: Part 8
    ; https://www.hivemq.com/blog/mqtt-essentials-part-8-retained-messages/

    Last Will and Testament - MQTT Essentials: Part 9
    ; https://www.hivemq.com/blog/mqtt-essentials-part-9-last-will-and-testament/

    Keep Alive and Client Take-Over - MQTT Essentials Part 10
    ; https://www.hivemq.com/blog/mqtt-essentials-part-10-alive-client-take-over/

그 외, 기타.
ESP8266 and Node-RED with MQTT (Publish and Subscribe)
; https://randomnerdtutorials.com/esp8266-and-node-red-with-mqtt/

Send and Receive Messages to your IoT Devices using MQTT
; https://www.digikey.com/en/maker/projects/send-and-receive-messages-to-your-iot-devices-using-mqtt/39ed5690cc46473abe8904c8f960341f




[이 글에 대해서 여러분들과 의견을 공유하고 싶습니다. 틀리거나 미흡한 부분 또는 의문 사항이 있으시면 언제든 댓글 남겨주십시오.]

[연관 글]


donaricano-btn



[최초 등록일: ]
[최종 수정일: 7/14/2021

Creative Commons License
이 저작물은 크리에이티브 커먼즈 코리아 저작자표시-비영리-변경금지 2.0 대한민국 라이센스에 따라 이용하실 수 있습니다.
by SeongTae Jeong, mailto:techsharer at outlook.com

비밀번호

댓글 쓴 사람
 



2021-06-05 11시51분
[사용자] 시간되실때 Prism 프레임워크 강의 좀 해주시면 안되나요? 국내에서는 WPF와 함께 가장 많이 사용되기는한데 처음 접하는 사람 입장에서는 도대체가 이해가 안가네요. 공식 사이트나 유튜브를 봐도 뭔지 모르겠고, 본문에서 언급한 Event Subscriber/Publisher 방법도 모르겠고...
[손님]
2021-06-07 11시07분
제가 WPF를 다룬 것이 거의 12년 정도 전이라서 이제는 어떻게 바뀌었는지도 잘 모릅니다. ^^ 다른 현업에 계신 분들이 더 실력이 좋기 때문에 그분들의 강의를 기대하는 것이 더 좋을 듯합니다.
정성태
2021-06-18 11시32분
Show dotnet: Running my .NET nanoFramework for 8 years on a battery
; https://devblogs.microsoft.com/dotnet/show-dotnet-running-my-net-nanoframework-for-8-years-on-a-battery/
정성태

1  2  3  4  5  6  7  8  9  10  11  12  [13]  14  15  ...
NoWriterDateCnt.TitleFile(s)
12540정성태2/17/2021851.NET Framework: 1024. C# - Win32 API에 대한 P/Invoke를 대신하는 Microsoft.Windows.CsWin32 패키지
12539정성태2/16/2021923Windows: 189. WM_TIMER의 동작 방식 개요파일 다운로드1
12538정성태2/15/20211055.NET Framework: 1023. C# - GC 힙이 아닌 Native 힙에 인스턴스 생성 - 0SuperComicLib.LowLevel 라이브러리 소개 [2]
12537정성태2/11/20211167.NET Framework: 1022. UI 요소의 접근은 반드시 그 UI를 만든 스레드에서! - 두 번째 이야기
12536정성태2/9/20211043개발 환경 구성: 542. BDP(Bandwidth-delay product)와 TCP Receive Window
12535정성태2/9/2021733개발 환경 구성: 541. Wireshark로 확인하는 LSO(Large Send Offload), RSC(Receive Segment Coalescing) 옵션
12534정성태2/8/2021999개발 환경 구성: 540. Wireshark + C/C++로 확인하는 TCP 연결에서의 closesocket 동작 [1]파일 다운로드1
12533정성태2/8/2021961개발 환경 구성: 539. Wireshark + C/C++로 확인하는 TCP 연결에서의 shutdown 동작파일 다운로드1
12532정성태2/6/2021873개발 환경 구성: 538. Wireshark + C#으로 확인하는 ReceiveBufferSize(SO_RCVBUF), SendBufferSize(SO_SNDBUF) [1]
12531정성태2/5/2021733개발 환경 구성: 537. Wireshark + C#으로 확인하는 PSH flag와 Nagle 알고리듬파일 다운로드1
12530정성태2/4/20211168개발 환경 구성: 536. Wireshark + C#으로 확인하는 TCP 통신의 Receive Window
12529정성태2/4/2021821개발 환경 구성: 535. Wireshark + C#으로 확인하는 TCP 통신의 MIN RTO [1]
12528정성태2/1/2021875개발 환경 구성: 534. Wireshark + C#으로 확인하는 TCP 통신의 MSS(Maximum Segment Size) - 윈도우 환경
12527정성태2/1/2021898개발 환경 구성: 533. Wireshark + C#으로 확인하는 TCP 통신의 MSS(Maximum Segment Size) - 리눅스 환경파일 다운로드1
12526정성태2/1/2021718개발 환경 구성: 532. Azure Devops의 파이프라인 빌드 시 snk 파일 다루는 방법 - Secure file
12525정성태2/1/2021606개발 환경 구성: 531. Azure Devops - 파이프라인 실행 시 빌드 이벤트를 생략하는 방법
12524정성태1/31/2021754개발 환경 구성: 530. 기존 github 프로젝트를 Azure Devops의 빌드 Pipeline에 연결하는 방법 [1]
12523정성태1/31/2021750개발 환경 구성: 529. 기존 github 프로젝트를 Azure Devops의 Board에 연결하는 방법
12522정성태1/31/2021901개발 환경 구성: 528. 오라클 클라우드의 리눅스 VM - 9000 MTU Jumbo Frame 테스트
12521정성태1/31/20211036개발 환경 구성: 527. 이더넷(Ethernet) 환경의 TCP 통신에서 MSS(Maximum Segment Size) 확인
12520정성태1/30/2021634개발 환경 구성: 526. 오라클 클라우드의 VM에 ping ICMP 여는 방법
12519정성태1/30/2021754개발 환경 구성: 525. 오라클 클라우드의 VM을 외부에서 접근하기 위해 포트 여는 방법
12518정성태1/30/20211682Linux: 37. Ubuntu에 Wireshark 설치
12517정성태1/30/20211324Linux: 36. 윈도우 클라이언트에서 X2Go를 이용한 원격 리눅스의 GUI 접속 - 우분투 20.04
12516정성태1/29/2021831Windows: 188. Windows - TCP default template 설정 방법
12515정성태1/28/2021934웹: 41. Microsoft Edge - localhost에 대해 http 접근 시 무조건 https로 바뀌는 문제 [3]
1  2  3  4  5  6  7  8  9  10  11  12  [13]  14  15  ...