Azure 의 EventHub를 셋팅하고  .Net Core Akka.net 액터모델을 만들고 기능을 연결해보겠습니다.

EventHub를 사용하는 전반적인 느낌은, KAFKA의 기능을 흡수한 KAFKA-SASS 였으며 

KAFKA의 셋팅과 토픽관리를 Azure 포털로 간편하게 할수 있었습니다.

사용프레임워크

• net core 3.1
• akka.net
• Confluent.Kafka

자바의 경우 다음 링크 AkkaStream With EventHub 에서 AKKAStream을 사용하여  연결가능하며

Akka.net에는 아직 EventHub(Kafka 모드)와 연결지원하는 모드를 찾지못하여

액터모델에 Kafka클라이언트를 탑재하여, 메시징 처리에유연한 액터모델버전으로 확장하였습니다.

BootstrapServers = "webnori-kafka.servicebus.windows.net:9093"

생성후 KAFKASUFACE가 활성화되는지 체크합니다.

( 저의경우 기본플랜으로 선택하였다가 리전변경및 여러번 실패후, 상위 플랜으로 재생성하여 활성화가 되었습니다.  -모든 리전지원합니다.   )

모든리전 지원하며, 기본플랜이아닌 표준플랜이상 KAFKA 인터페이스가 지원합니다.

양끝단을 Kafka 인터페이스로 맞추고, 클라우드로 관리되는 큐로 보여집니다.

            var producerAkkaOption = new ProducerAkkaOption()
            {
                BootstrapServers = "webnori-kafka.servicebus.windows.net:9093",
                ProducerName = "webnori-kafka",
                SecuritOption = new KafkaSecurityOption()
                {
                    SecurityProtocol = SecurityProtocol.SaslSsl,
                    SaslMechanism = SaslMechanism.Plain,
                    SaslUsername = "$ConnectionString",
                    SaslPassword = "Endpoint=sb://webnori-kafka.servicebus.windows.net/;SharedAccessKeyName=kafka-client;SharedAccessKey=PfL0qRUm50AXZHRXLiVfnatIRI3OqAh+dT6Owsqrd2M=",
                    SslCaLocation = "./cacert.pem"
                }
            };

            var producerAkkaOption = new ProducerAkkaOption()
            {
                BootstrapServers = "webnori-kafka.servicebus.windows.net:9093",
                ProducerName = "webnori-kafka",
                SecuritOption = new KafkaSecurityOption()
                {
                    SecurityProtocol = SecurityProtocol.SaslSsl,
                    SaslMechanism = SaslMechanism.Plain,
                    SaslUsername = "$ConnectionString",
                    SaslPassword = "Endpoint=sb://webnori-kafka.servicebus.windows.net/;SharedAccessKeyName=kafka-client;SharedAccessKey=PfL0qRUm50AXZHRXLiVfnatIRI3OqAh+dT6Owsqrd2M=",
                    SslCaLocation = "./cacert.pem"
                }
            };

            string producerActorName = "producerActor";

            var producerActor= AkkaLoad.RegisterActor(producerActorName /*AkkaLoad가 인식하는 유니크명*/,
                actorSystem.ActorOf(Props.Create(() => 
                    new ProducerActor(producerAkkaOption)),
                    producerActorName /*AKKA가 인식하는 Path명*/
            ));

            producerActor.Tell(new BatchData()
            {
                Data = new KafkaTextMessage()
                {
                    Topic = "akka100",
                    Message = "testData"
                }
            });

            //ConsumerActor
            var consumerAkkaOption = new ConsumerAkkaOption()
            {
                BootstrapServers = "webnori-kafka.servicebus.windows.net:9093",
                Topics = "akka100",
                AutoOffsetReset = AutoOffsetReset.Earliest,
                KafkaGroupId = "akakTestGroup",
                RelayActor = null,  //작업자 액터를 연결하면, 소비메시지가 작업자에게 전달된다 ( 컨슘기능과 작업자 기능의 분리)
                SecurityOption = new KafkaSecurityOption()
                {
                    SecurityProtocol = SecurityProtocol.SaslSsl,
                    SaslMechanism = SaslMechanism.Plain,
                    SaslUsername = "$ConnectionString",
                    SaslPassword = "Endpoint=sb://webnori-kafka.servicebus.windows.net/;SharedAccessKeyName=kafka-client;SharedAccessKey=PfL0qRUm50AXZHRXLiVfnatIRI3OqAh+dT6Owsqrd2M=",
                    SslCaLocation = "./cacert.pem"
                }
            };

            string consumerActorName = "consumerActor";
            var consumerActor = AkkaLoad.RegisterActor(consumerActorName /*AkkaLoad가 인식하는 유니크명*/,
                actorSystem.ActorOf(Props.Create(() =>
                    new ConsumerActor(consumerAkkaOption)),
                    consumerActorName /*AKKA가 인식하는 Path명*/
            ));

            //컨슈머를 작동시킨다.
            consumerActor.Tell(new ConsumerStart());

<PackageReference Include="Confluent.Kafka" Version="1.5.2" />