基於Confluent.Kafka實現的Kafka客戶端操作類使用詳解

一、引言

    　　有段時間沒有寫東西了，當然不是沒得寫，還有MongoDB的系列沒有寫完呢，那個系列還要繼續。今天正好是周末，有點時間，來寫新東西吧。最近公司用了Kafka做為消息的中間件，最開始寫的那個版本不是很好，我就要來優化它，所以就抽了一些時間來研究Kafka。很多概念性的東西就不寫了，今天主要是上干貨，主要是代碼，今天就把Kafka的消費者和生產者的代碼貼出來，以供大家參考，當然這個是代碼樣板，最后我也會把地址貼出來。以后有時間我會把我自己實現的Kafka消息的生產者和消費者的代碼貼出來。好了，話不多說，言歸正傳。

    　　說明一點，如果想調試這里的代碼，必須引入Confluent.Kafka這個dll才可以，直接在Visual Studio 項目的 Nuget 里面可以查找，直接安裝就可以了。

二、消息的生產者（Kafka消息的Producer）

    　　大多數的消息中間件都包含三個部分，一個是消息的生產者，一個是存放消息的隊列，另外一個就是消息的消費者，我們就按着這個順序，我就先把消息生產者的代碼寫出來。直接上代碼，其實不是很難，里面有很多備注，只要有基本的概念理解起來還是很容易的。

　　   第一個版本，同步版本！

using System;
using System.IO;
using System.Text;
using System.Collections.Generic;
using Confluent.Kafka;
using Confluent.Kafka.Serialization;


namespace Confluent.Kafka.Examples.Producer
{
    public class Program
    {
        public static void Main(string[] args)
        {
            if (args.Length != 2)
            {
                Console.WriteLine("Usage: .. brokerList topicName");
                return;
            }

            string brokerList = args[0];
            string topicName = args[1];

            var config = new Dictionary<string, object> { { "bootstrap.servers", brokerList } };

            using (var producer = new Producer<string, string>(config, new StringSerializer(Encoding.UTF8), new StringSerializer(Encoding.UTF8)))
            {
                var cancelled = false;
                Console.CancelKeyPress += (_, e) => {
                    e.Cancel = true; // 阻止進程退出
                    cancelled = true;
                };

                while (!cancelled)
                {
                    Console.Write("> ");

                    string text;
                    try
                    {
                        text = Console.ReadLine();
                    }
                    catch (IOException)
                    {
                        // IO 異常拋出的時候設置此值ConsoleCancelEventArgs.Cancel == true.
                        break;
                    }
                    if (text == null)
                    {
                        break;
                    }

                    string key = null;
                    string val = text;

                    // 如果指定了鍵和值，則拆分行.
                    int index = text.IndexOf(" ");
                    if (index != -1)
                    {
                        key = text.Substring(0, index);
                        val = text.Substring(index + 1);
                    }

                    // 在下面的異步生產請求上調用.Result會導致它阻塞，直到它完成。 通常，您應該避免同步生成，因為這會對吞吐量產生巨大影響。對於這個交互式控制台的例子，這是我們想要的。
                    var deliveryReport = producer.ProduceAsync(topicName, key, val).Result;
                    Console.WriteLine(
                        deliveryReport.Error.Code == ErrorCode.NoError
                            ? "delivered to: "+deliveryReport.TopicPartitionOffset
                            : "failed to deliver message: "+deliveryReport.Error.Reason
                    );
                }

                // 由於我們是同步的生產消息，此時不會有消息在傳輸並且也不需要等待消息到達的確認應答, 銷毀生產者之前我們是不需要調用 producer.Flush 方法, 就像正常使用一樣。
            }
        }
    }
}

 

　　　　第二個版本，異步版本，推薦使用

using System;
using System.IO;
using System.Text;
using System.Collections.Generic;
using Confluent.Kafka;
using Confluent.Kafka.Serialization;


namespace Confluent.Kafka.Examples.Producer
{
    public class Program
    {
        public static void Main(string[] args)
        {
            if (args.Length != 2)
            {
                Console.WriteLine("Usage: .. brokerList topicName");
                return;
            }

            string brokerList = args[0];
            string topicName = args[1];
            string message="我就是要傳輸的消息內容";

            //這是以異步方式生產消息的代碼實例
            var config = new Dictionary<string, object> { { "bootstrap.servers", brokerList } };
            using (var producer = new Producer<Null, string>(config, null, new StringSerializer(Encoding.UTF8)))
            {
                var deliveryReport = producer.ProduceAsync(topicName, null, message);
                deliveryReport.ContinueWith(task =>
                {
                    Console.WriteLine("Producer: "+producer.Name+"\r\nTopic: "+topicName+"\r\nPartition: "+task.Result.Partition+"\r\nOffset: "+task.Result.Offset);
                });
 
                producer.Flush(TimeSpan.FromSeconds(10));
           }
        }
    }
}

    　　好了，上面給出了兩個版本的消息生產者的代碼，一個是同步版本，第二個是異步版本的，推薦使用異步版本的代碼實現。

三、消息的消費者（Kafka消息的Consumer）

    　　在消息的生產者中已經說明了消息中間件的三個部分，第一個是消息的生產者，沒有消息的生產者，就沒有消息的消費者了，巧婦難為無米之炊吧。在上一節我們已經寫了消息生產者的代碼，這一節，我們主要來貼出消息消費者的代碼。代碼同樣很簡單，注釋也很全。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Confluent.Kafka.Serialization;


/// <summary>
///     演示如何使用Consumer客戶端.
/// </summary>
namespace Confluent.Kafka.Examples.Consumer
{
    public class Program
    {
        /// <summary>
        //      在這個例子中:
        ///         - offsets 是自動提交的。
        ///         - consumer.Poll / OnMessage 是用於消息消費的。
        ///         - 沒有為輪詢循環創建（Poll）二外的線程，當然可以創建
        /// </summary>
        public static void Run_Poll(string brokerList, List<string> topics)
        {
            var config = new Dictionary<string, object>
            {
                { "bootstrap.servers", brokerList },
                { "group.id", "csharp-consumer" },
                { "enable.auto.commit", true },  // 默認值
                { "auto.commit.interval.ms", 5000 },
                { "statistics.interval.ms", 60000 },
                { "session.timeout.ms", 6000 },
                { "auto.offset.reset", "smallest" }
            };

            using (var consumer = new Consumer<Ignore, string>(config, null, new StringDeserializer(Encoding.UTF8)))
            {
                // 注意: 所有事件處理程序的執行都是在主線程中執行的，就是同步的。

                //當成功消費了消息就會觸發該事件
                consumer.OnMessage += (_, msg) => Console.WriteLine("Topic: "+msg.Topic+" Partition: "+msg.Partition+" Offset: "+msg.Offset+" "+msg.Value);

                consumer.OnPartitionEOF += (_, end) => Console.WriteLine("Reached end of topic "+end.Topic+" partition "+end.Partition+", next message will be at offset "+end.Offset);

                //當然發生了嚴重錯誤，比如，連接丟失或者Kafka服務器無效就會觸發該事件
                consumer.OnError += (_, error) => Console.WriteLine("Error: "+error);

                //當反序列化有錯誤，或者消費的過程中發生了錯誤，即error != NoError，就會觸發該事件
                consumer.OnConsumeError += (_, msg)
                    => Console.WriteLine("Error consuming from topic/partition/offset "+msg.Topic+"/"+msg.Partition+"/"+msg.Offset+": "+msg.Error);

                //成功提交了Offsets會觸發該事件
                consumer.OnOffsetsCommitted += (_, commit) => Console.WriteLine(commit.Error ? "Failed to commit offsets: "+commit.Error : "Successfully committed offsets: "+commit.Offsets);

                // 當消費者被分配一組新的分區時觸發該事件
                consumer.OnPartitionsAssigned += (_, partitions) =>
                {
                    Console.WriteLine("Assigned partitions:"+partitions+"  "+member id: "+consumer.MemberId);
                    // 如果您未向OnPartitionsAssigned事件添加處理程序，則會自動執行以下.Assign調用。 如果你這樣做，你必須明確地調用.Assign以便消費者開始消費消息。
                    //開始從分區中消息消息
                    consumer.Assign(partitions);
                };

                // 當消費者的當前分區集已被撤銷時引發該事件。
                consumer.OnPartitionsRevoked += (_, partitions) =>
                {
                    Console.WriteLine("Revoked partitions:"+partitions);
                    // 如果您未向OnPartitionsRevoked事件添加處理程序，則下面的.Unassign調用會自動發生。 如果你這樣做了，你必須明確地調用.Usessign以便消費者停止從它先前分配的分區中消費消息。

                    //停止從分區中消費消息
                    consumer.Unassign();
                };

                consumer.OnStatistics += (_, json) => Console.WriteLine("Statistics: "+json);

                consumer.Subscribe(topics);

                Console.WriteLine("Subscribed to:"+consumer.Subscription);

                var cancelled = false;
                Console.CancelKeyPress += (_, e) => {
                    e.Cancel = true;  // 組織進程退出
                    cancelled = true;
                };

                Console.WriteLine("Ctrl-C to exit.");
                while (!cancelled)
                {
                    consumer.Poll(TimeSpan.FromMilliseconds(100));
                }
            }
        }

        /// <summary>
        ///     在這實例中
        ///         - offsets 是手動提交的。
        ///         - consumer.Consume方法用於消費消息
        ///             (所有其他事件仍由事件處理程序處理)
        ///         -沒有為了 輪詢（消耗）循環 創建額外的線程。
        /// </summary>
        public static void Run_Consume(string brokerList, List<string> topics)
        {
            var config = new Dictionary<string, object>
            {
                { "bootstrap.servers", brokerList },
                { "group.id", "csharp-consumer" },
                { "enable.auto.commit", false },
                { "statistics.interval.ms", 60000 },
                { "session.timeout.ms", 6000 },
                { "auto.offset.reset", "smallest" }
            };

            using (var consumer = new Consumer<Ignore, string>(config, null, new StringDeserializer(Encoding.UTF8)))
            {
                // 注意:所有事件處理都是在主線程中處理的，也就是說同步的

                consumer.OnPartitionEOF += (_, end)
                    => Console.WriteLine("Reached end of topic "+end.Topic+" partition "+end.Partition+", next message will be at offset "+end.Offset);

                consumer.OnError += (_, error)=> Console.WriteLine("Error: "+error);

                // 當反序列化有錯誤，或者消費的過程中發生了錯誤，即error != NoError，就會觸發該事件
                consumer.OnConsumeError += (_, error)=> Console.WriteLine("Consume error: "+error);

                // 當消費者被分配一組新的分區時觸發該事件
                consumer.OnPartitionsAssigned += (_, partitions) =>
                {
                    Console.WriteLine("Assigned partitions:"+partitions+"  "+member id: "+consumer.MemberId);
                    // 如果您未向OnPartitionsAssigned事件添加處理程序，則會自動執行以下.Assign調用。 如果你這樣做，你必須明確地調用.Assign以便消費者開始消費消息。
                    //開始從分區中消息消息
                    consumer.Assign(partitions);
                };

                // 當消費者的當前分區集已被撤銷時引發該事件。
                consumer.OnPartitionsRevoked += (_, partitions) =>
                {
                    Console.WriteLine("Revoked partitions:"+partitions);
                    // 如果您未向OnPartitionsRevoked事件添加處理程序，則下面的.Unassign調用會自動發生。 如果你這樣做了，你必須明確地調用.Usessign以便消費者停止從它先前分配的分區中消費消息。

                    //停止從分區中消費消息
                    consumer.Unassign();
                };

                consumer.OnStatistics += (_, json) => Console.WriteLine("Statistics: "+json);

                consumer.Subscribe(topics);

                Console.WriteLine("Started consumer, Ctrl-C to stop consuming");

                var cancelled = false;
                Console.CancelKeyPress += (_, e) => {
                    e.Cancel = true; // 防止進程退出
                    cancelled = true;
                };

                while (!cancelled)
                {
                    if (!consumer.Consume(out Message<Ignore, string> msg, TimeSpan.FromMilliseconds(100)))
                    {
                        continue;
                    }

                    Console.WriteLine("Topic: "+msg.Topic+" Partition: "+msg.Partition+" Offset: "+msg.Offset+" "+msg.Value);

                    if (msg.Offset % 5 == 0)
                    {
                        var committedOffsets = consumer.CommitAsync(msg).Result;
                        Console.WriteLine("Committed offset: "+committedOffsets);
                    }
                }
            }
        }

        /// <summary>
        ///     在這個例子中
        ///         - 消費者組功能（即.Subscribe +offset提交）不被使用。
        ///         - 將消費者手動分配給分區，並始終從特定偏移量（0）開始消耗。
        /// </summary>
        public static void Run_ManualAssign(string brokerList, List<string> topics)
        {
            var config = new Dictionary<string, object>
            {
                // 即使您不打算使用任何使用者組功能，也必須在創建使用者時指定group.id屬性。
                { "group.id", new Guid().ToString() },
                { "bootstrap.servers", brokerList },
                // 即使消費者沒有訂閱該組，也可以將分區偏移量提交給一個組。 在這個例子中，自動提交被禁用以防止發生這種情況。
                { "enable.auto.commit", false }
            };

            using (var consumer = new Consumer<Ignore, string>(config, null, new StringDeserializer(Encoding.UTF8)))
            {
                //總是從0開始消費
                consumer.Assign(topics.Select(topic => new TopicPartitionOffset(topic, 0, Offset.Beginning)).ToList());

                // 引發嚴重錯誤，例如 連接失敗或所有Kafka服務器失效。
                consumer.OnError += (_, error) => Console.WriteLine("Error: "+error);

                // 這個事件是由於在反序列化出現錯誤，或者在消息消息的時候出現錯誤，也就是 error != NoError 的時候引發該事件
                consumer.OnConsumeError += (_, error) => Console.WriteLine("Consume error: "+error);

                while (true)
                {
                    if (consumer.Consume(out Message<Ignore, string> msg, TimeSpan.FromSeconds(1)))
                    {
                        Console.WriteLine("Topic: "+msg.Topic+" Partition: "+msg.Partition+" Offset: "+msg.Offset+" "+msg.Value);
                    }
                }
            }
        }

        private static void PrintUsage()=> Console.WriteLine("Usage: .. <poll|consume|manual> <broker,broker,..> <topic> [topic..]");

        public static void Main(string[] args)
        {
            if (args.Length < 3)
            {
                PrintUsage();
                return;
            }

            var mode = args[0];
            var brokerList = args[1];
            var topics = args.Skip(2).ToList();

            switch (mode)
            {
                case "poll":
                    Run_Poll(brokerList, topics);
                    break;
                case "consume":
                    Run_Consume(brokerList, topics);
                    break;
                case "manual":
                    Run_ManualAssign(brokerList, topics);
                    break;
                default:
                    PrintUsage();
                    break;
            }
        }
    }
}

    　　以上代碼也有兩個版本，第一個版本是自動提交Offset，第二個版本是人工提交Offset，但是代碼沒有分開寫，只是不同的版本用了不同的方法。    

四、結束

    　　好了，今天就寫到這里了，這是一個引子，所有代碼都是真實有效的，我已經全部測試過，所以大家可以放心使用或者改造成自己的消息的生產者和消息消費者的實現。原文的地址如下，https://github.com/confluentinc/confluent-kafka-dotnet/tree/master/examples    ，內容差不多。不忘初心，繼續努力吧。