.net windows Kafka 安裝與使用入門(入門筆記)，.netkafka

.net windows Kafka 安裝與使用入門(入門筆記)，.netkafka

完整解決方案請參考：

Setting Up and Running Apache Kafka on Windows OS

 

在環境搭建過程中遇到兩個問題，在這裡先列出來，以方便查詢：

1. \Java\jre7\lib\ext\QTJava.zip was unexpected at this time. Process exited 解決方案： 1.1 右鍵點擊“我的電腦” -> "高級系統設置" -> "環境變量" 1.2 查看CLASSPATH的值裡有沒有包含QTJava.zip的路徑，如果有的話，把對應的路徑刪除，問題就解決了。   2. Missing `server' JVM (Java\jre7\bin\server\jvm­­.dll.) 解決方案： 2.1. 拷貝C:\Program Files\Java\jdk1.6.0\jre\bin\server 2.2 粘貼到 C:\Program Files\Java\jre1.6.0\bin  

搭建環境

1. 安裝JDK

1.1 安裝文件：http://www.oracle.com/technetwork/java/javase/downloads/jre8-downloads-2133155.html下載Server JRE.
1.2 安裝完成後需要添加以下的環境變量（右鍵點擊“我的電腦” -> "高級系統設置" -> "環境變量" ）：

• • JAVA_HOME: C:\Program Files (x86)\Java\jre1.8.0_60（這個是默認安裝路徑，如果安裝過程中更改了安裝目錄，把更改後的路徑填上就行了）

• • PATH: 在現有的值後面添加"; %JAVA_HOME%\bin"

1.3 打開cmd運行 "java -version" 查看當前系統Java的版本：

 

2. 安裝Zookeeper

Kafka的運行依賴於Zookeeper，所以在運行Kafka之前我們需要安裝並運行Zookeeper

2.1 下載安裝文件： http://zookeeper.apache.org/releases.html 2.2 解壓文件（本文解壓到 G:\zookeeper-3.4.8） 2.3 打開G:\zookeeper-3.4.8\conf，把zoo_sample.cfg重命名成zoo.cfg 2.4 從文本編輯器裡打開zoo.cfg 2.5 把dataDir的值改成“:\zookeeper-3.4.8\data” 2.6 添加如下系統變量：

• • ZOOKEEPER_HOME: G:\zookeeper-3.4.8
  • Path: 在現有的值後面添加 ";%ZOOKEEPER_HOME%\bin;"

2.7 運行Zookeeper: 打開cmd然後執行

zkserver

3. 安裝並運行Kafka

3.1 下載安裝文件： http://kafka.apache.org/downloads.html 3.2 解壓文件（本文解壓到 G:\kafka_2.11-0.10.0.1） 3.3 打開G:\kafka_2.11-0.10.0.1\config 3.4 從文本編輯器裡打開 server.properties 3.5 把 log.dirs的值改成 “G:\kafka_2.11-0.10.0.1\kafka-logs” 3.6 打開cmd 3.7 進入kafka文件目錄: cd /d G:\kafka_2.11-0.10.0.1\ 3.8 輸入並執行以打開kafka:

.\bin\windows\kafka-server-start.bat .\config\server.properties

4. 創建topics

4.1 打開cmd 並進入G:\kafka_2.11-0.10.0.1\bin\windows 4.2 創建一個topic：

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

5. 打開一個Producer：

cd /d G:\kafka_2.11-0.10.0.1\bin\windows
kafka-console-producer.bat --broker-list localhost:9092 --topic test

 

6. 打開一個Consumer:

cd /d G:\kafka_2.11-0.10.0.1\bin\windows

kafka-console-consumer.bat --zookeeper localhost:2181 --topic test

然後就可以在Producer控制台窗口輸入消息了。在消息輸入過後，很快Consumer窗口就會顯示出Producer發送的消息：

 

至此，Kafka運行環境的搭建就完成了:-)

web管理方面（正式運維部署）：

網上有人說 kafka manage

.net sdk方面：

有人建議:https://github.com/Microsoft/CSharpClient-for-Kafka （微軟提供的sdk，看了源碼個人感覺有些復雜）

還有
kafka-net 這個感覺會簡單很多（看了源碼感覺一些Task寫法並不喜歡）

推薦:kafka-net 簡單易用些。

地址: https://github.com/Jroland/kafka-net

 

以上sdk 可以搜索nuget 查看開源地址和最新release包,當然.net sdk 還有很多其他選擇。

個人開發工具（windows支持），windows 開發環境使用

kafkaTool

下載地址: http://www.kafkatool.com/download.html

 

 

kafka-net sdk使用demo

參考來源:http://www.cnblogs.com/Wulex/p/5619425.html

詳細請查看官方開源地址的demo: https://github.com/Jroland/kafka-net

KafkaProducer程序: 

class Program
    {
        static void Main(string[] args)
        {
            do
            {
                Produce(GetKafkaBroker(), getTopicName());
                System.Threading.Thread.Sleep(3000);
            } while (true);
        }

        private static void Produce(string broker, string topic)
        {
            var options = new KafkaOptions(new Uri(broker));
            var router = new BrokerRouter(options);
            var client = new Producer(router);

            var currentDatetime =DateTime.Now;
            var key = currentDatetime.Second.ToString();
            var events = new[] { new Message("Hello World " + currentDatetime, key) };
            client.SendMessageAsync(topic, events).Wait(1500);
            Console.WriteLine("Produced: Key: {0}. Message: {1}", key, events[0].Value.ToUtf8String());

            using (client) { }
        }

        private static string GetKafkaBroker()
        {
            string KafkaBroker = string.Empty;
            const string kafkaBrokerKeyName = "KafkaBroker";

            if (!ConfigurationManager.AppSettings.AllKeys.Contains(kafkaBrokerKeyName))
            {
                KafkaBroker = "http://localhost:9092";
            }
            else
            {
                KafkaBroker = ConfigurationManager.AppSettings[kafkaBrokerKeyName];
            }
            return KafkaBroker;
        }
        private static string getTopicName()
        {
            string TopicName = string.Empty;
            const string topicNameKeyName = "Topic";

            if (!ConfigurationManager.AppSettings.AllKeys.Contains(topicNameKeyName))
            {
                throw new Exception("Key \"" + topicNameKeyName + "\" not found in Config file -> configuration/AppSettings");
            }
            else
            {
                TopicName = ConfigurationManager.AppSettings[topicNameKeyName];
            }
            return TopicName;
        }
    }

4.KafkaConsumer程序:

class Program
    {
        static void Main(string[] args)
        {
            Consume(getKafkaBroker(), getTopicName());
            
        }

        private static void Consume(string broker, string topic)
        {   
            var options = new KafkaOptions(new Uri(broker));
            var router = new BrokerRouter(options);
            var consumer = new Consumer(new ConsumerOptions(topic, router));

            //Consume returns a blocking IEnumerable (ie: never ending stream)
            foreach (var message in consumer.Consume())
            {
                Console.WriteLine("Response: Partition {0},Offset {1} : {2}",
                    message.Meta.PartitionId, message.Meta.Offset, message.Value.ToUtf8String());
            }
        }

        private static string getKafkaBroker()
        {
            string KafkaBroker = string.Empty;
            var KafkaBrokerKeyName = "KafkaBroker";

            if (!ConfigurationManager.AppSettings.AllKeys.Contains(KafkaBrokerKeyName))
            {
                KafkaBroker = "http://localhost:9092";
            }
            else
            {
                KafkaBroker = ConfigurationManager.AppSettings[KafkaBrokerKeyName];
            }
            return KafkaBroker;
        }

        private static string getTopicName()
        {
            string TopicName = string.Empty;
            var TopicNameKeyName = "Topic";

            if (!ConfigurationManager.AppSettings.AllKeys.Contains(TopicNameKeyName))
            {
                throw new Exception("Key \"" + TopicNameKeyName + "\" not found in Config file -> configuration/AppSettings");
            }
            else
            {
                TopicName = ConfigurationManager.AppSettings[TopicNameKeyName];
            }
            return TopicName;
        }
    }

5.Consumer結果: