Kafka .net 開發入門

　　Kafka安裝

　　首先我們需要在windows服務器上安裝kafka以及zookeeper,有關zookeeper的介紹將會在后續進行講解。

       在網上可以找到相應的安裝方式，我采用的是騰訊雲服務器，借鑒的是https://www.cnblogs.com/lnice/p/9668750.html

       根據上面博客安裝完成后，我們在kafka中新建了一個名叫test的Topic,並新建了一個生產者和一個消費者。

　　注：控制台生產者和控制台消費者的數據不同是因為我用.net開發了一個生產者所致，后面會講

　　

　　.net生產者　　

　　當前市面上比較好的.NET的kafka開源包有兩個：kafka-net和rdkafka,我采用的是RdKafka；

　　新建項目后首先添加Nuget包，我的生產者源碼如下所示

            string brokerList = "118.24.184.36:9092";
            string topicName = "test";


            using (Producer producer = new Producer("118.24.184.36:9092"))
            using (Topic topic = producer.Topic(topicName))
            {
                Console.WriteLine("{" + producer.Name + "} producing on {" + topic.Name + "}. q to exit.");

                string text;
                while ((text = Console.ReadLine()) != "q")
                {
                    byte[] data = Encoding.UTF8.GetBytes(text);
                    Task<DeliveryReport> deliveryReport = topic.Produce(data);
                    var unused = deliveryReport.ContinueWith(task =>
                    {
                        Console.WriteLine("Partition: {" + task.Result.Partition + "}, Offset: {" + task.Result.Offset + "}");
                    });
                 }
            }
            

　　如你所見，當我們在軟件的輸入框輸入hellow world並回車后，控制台的消費者就收到了。

　　

　　如上圖，控制台消費者收到了.net生產者發布的消息。

　　

　　.net 消費者

　　我采用的還是RfKafka這個插件，采用的代碼如下

 

static void Main(string[] args)
        {
            string brokerList = "118.24.184.36:9092";
            List<string> topic = new List<string>();
            topic.Add("test");
            Run(brokerList, topic);
        }
        public static void Run(string brokerList, List<string> topics)
        {
            bool enableAutoCommit = false;

            var config = new Config()
            {
                GroupId = "advanced-csharp-consumer",
                EnableAutoCommit = enableAutoCommit,
                StatisticsInterval = TimeSpan.FromSeconds(60)
            };

            using (var consumer = new EventConsumer(config, brokerList))
            {
                consumer.OnMessage += (obj, msg) =>
                {
                    string text = Encoding.UTF8.GetString(msg.Payload, 0, msg.Payload.Length);
                    Console.WriteLine($"Topic: {msg.Topic} Partition: {msg.Partition} Offset: {msg.Offset} {text}");
                    Console.WriteLine("1 Response: Partition {0},Offset {1} : {2}",
                    msg.Partition, msg.Offset, text);
                    if (!enableAutoCommit && msg.Offset % 10 == 0)
                    {
                        Console.WriteLine("Committing offset");
                        consumer.Commit(msg).Wait();
                        Console.WriteLine("Committed offset");
                    }
                };

                consumer.OnConsumerError += (obj, errorCode) =>
                {
                    Console.WriteLine($"Consumer Error: {errorCode}");
                };

                consumer.OnEndReached += (obj, end) =>
                {
                    Console.WriteLine($"Reached end of topic {end.Topic} partition {end.Partition}, next message will be at offset {end.Offset}");
                };

                consumer.OnError += (obj, error) =>
                {
                    Console.WriteLine($"Error: {error.ErrorCode} {error.Reason}");
                };

                if (enableAutoCommit)
                {
                    consumer.OnOffsetCommit += (obj, commit) =>
                    {
                        if (commit.Error != ErrorCode.NO_ERROR)
                        {
                            Console.WriteLine($"Failed to commit offsets: {commit.Error}");
                        }
                        Console.WriteLine($"Successfully committed offsets: [{string.Join(", ", commit.Offsets)}]");
                    };
                }

                consumer.OnPartitionsAssigned += (obj, partitions) =>
                {
                    Console.WriteLine($"Assigned partitions: [{string.Join(", ", partitions)}], member id: {consumer.MemberId}");
                    consumer.Assign(partitions);
                };

                consumer.OnPartitionsRevoked += (obj, partitions) =>
                {
                    Console.WriteLine($"Revoked partitions: [{string.Join(", ", partitions)}]");
                    consumer.Unassign();
                };

                consumer.OnStatistics += (obj, json) =>
                {
                    Console.WriteLine($"Statistics: {json}");
                };

                consumer.Subscribe(topics);
                consumer.Start();

                Console.WriteLine($"Assigned to: [{string.Join(", ", consumer.Assignment)}]");
                Console.WriteLine($"Subscribed to: [{string.Join(", ", consumer.Subscription)}]");

                Console.WriteLine($"Started consumer, press enter to stop consuming");
                Console.ReadLine();
            }
        }

　　運行后得到如下效果，以下四個框分別是.net生產者，控制台生產者，.net消費者，控制台消費者。

　　

　　至此.net關於kafka的使用就講完了。

 

       這里有一篇文章分享給大家，把很多kafka的一些深入理解進行了很通俗的講解：

http://melanx.com/2019/01/07/深入淺出理解基於-kafka-和-zookeeper-的分布式消息隊列/#12kafkatopic

 

　　如果你在使用上訴代碼時遇到了如下問題：

解決方案如下：將工程切到.net4.0,再將工程切到.net4.5或你需要的版本

　　不要問我為什么，我是這樣解決的，否則就需要根據缺失的庫進行相關插件安裝。