技術文章 » Kafka部署安裝

Kafka部署安裝

2022-08-08 06:04:52

一、環境準備

1、jdk 8+

2、zookeeper 

3、kafka

說明：在kafka較新版本中已經整合了zookeeper，所以不用單獨安裝zookeeper，只需要在kafka檔案目錄中啟動zookeeper即可

二、下載地址

https://kafka.apache.org/downloads

 

 

三、部署

1、啟動zookeeper

-- 啟動
./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

-- 檢視是否啟動成功
ps -ef | grep zoo

2、進入解壓的kafka目錄，修改/config/kafka-server的組態檔

vi config/server.properties
-- 重點設定節點說明

listeners=PLAINTEXT://localhost:9092   -- 將localhost修改為主機ip

log.dirs=/bigdata/kafka/logs-1  -- 預設不修改也可

 

 3、使用kafka-server-start.sh，啟動kafka服務

./bin/kafka-server-start.sh config/server.properties

四、使用使用者端kafka tools連線kafka

使用者端下載地址：https://www.kafkatool.com/download.html

關於使用者端如何使用可檢視：https://www.cnblogs.com/frankdeng/p/9452982.html

 

 

 

 

 

 

 

 

五、kafka實戰簡單使用

 

NuGet:Confluent.Kafka

 1、新建.Net Core控制檯專案，程式碼如下：

 

 

static void Main(string[] args)
        {
            // 傳送訊息
            var producerConfig = new ProducerConfig
            {
                BootstrapServers = "192.168.140.131:9092",
                MessageTimeoutMs = 50000
            };
            var builder = new ProducerBuilder<string, string>(producerConfig);
            using (var producer = builder.Build())
            {
                var data = new { key = "1", value = "001" };
                var json = JsonConvert.SerializeObject(data);
                var dr = producer.ProduceAsync("order", new Message<string, string> { Key = "order", Value = json }).GetAwaiter().GetResult();
                Console.WriteLine($"傳送事件{dr.Value}到{dr.TopicPartitionOffset}成功");
            }

            // 消費訊息
            var consumerConfig = new ConsumerConfig {
                BootstrapServers = "192.168.140.131:9092",
                AutoOffsetReset=AutoOffsetReset.Earliest,
                GroupId="1111", // 自定義
                EnableAutoCommit=true
            };
            var consumerBuilder = new ConsumerBuilder<string, string>(consumerConfig);


            using (var consumer = consumerBuilder.Build())
            {
                // 1、訂閱
                consumer.Subscribe("order");
                while (true)
                {
                    try
                    {
                        // 2、消費(自動確認)
                        var result = consumer.Consume();

                        // 3、業務邏輯
                        string key = result.Key;
                        string value = result.Value;

                        Console.WriteLine($"建立商品：Key:{key}");
                        Console.WriteLine($"建立商品：Order:{value}");
                        consumer.Commit(result);

                    }
                    catch (Exception e)
                    {
                        Console.WriteLine($"異常：Order:{e}");
                    }
                }
            }
        }

 

2、使用kafka使用者端檢視訊息投遞