前言：基于Windows系統下的Kafka環境搭建；以及使用.NET 6環境進行開發簡單的生産者與消費者的示範。
一、環境部署
Kafka是使用Java語言和Scala語言開發的，是以需要有對應的Java環境，以及Scala語言環境。
Java環境配置，如果不清楚的，可以檢視鄙人的另一篇部落格：
https://www.cnblogs.com/weskynet/p/14852471.html
1、Scala環境安裝，需要先下載下傳Scala語言包,下載下傳位址：
https://www.scala-lang.org/download/scala2.html
要選擇Binaries版本的環境，否則需要自己編譯：
2、Kafka基于Zookeeper環境運作，zookeeper提供給kafka一系列的功能支援，是以還需要安裝Zookeeper有關的環境。下載下傳zookeeper位址:
https://zookeeper.apache.org/releases.html#download
4、接下來是下載下傳主角，Kafka了。下載下傳位址：
https://kafka.apache.org/downloads.html
8、驗證Scala語言包是否安裝成功：
控制台視窗，輸入：scala -version
如果提示類似如下有關版本資訊，則代表安裝成功。
9、然後是安裝zookeeper環境。必須先啟動zookeeper，才可以使用kafka。
安裝zookeeper環境，先解壓下載下傳的包，然後在解壓後的目錄下新增data檔案夾
20、輸入指令：
.\bin\windows\kafka-server-start.bat .\config\server.properties
進行啟動Kafka服務：
26、然後是要一款Kafka可視化工具，此處我選擇使用offset explorer (原來是叫kafka tools，如下載下傳位址所示)，下載下傳位址：
https://www.kafkatool.com/download.html
二、代碼開發與測試
31、建立類庫項目，當作kafka服務類庫
/// <summary>
/// Description: Kafka生産者釋出服務
/// CreateTime: 2022/1/21 19:35:27
/// Author: Wesky
/// </summary>
public class PublishService: IPublishService
{
public async Task PublishAsync<TMessage>(string broker, string topicName, TMessage message) where TMessage : class
{
var config = new ProducerConfig
{
BootstrapServers = broker, // kafka服務叢集，例如 "192.168.0.1:9092,192.168.0.2:9092" 或者單機 "192.168.0.1:9092"
Acks = Acks.All,
MessageSendMaxRetries = 3, // 發送失敗重試的次數
};
using (var producer = new ProducerBuilder<string, string>(config).Build())
{
try
{
string data = Newtonsoft.Json.JsonConvert.SerializeObject(message);
var sendData = new Message<string, string> { Key = Guid.NewGuid().ToString("N"), Value = data};
var report = await producer.ProduceAsync(topicName, sendData);
Console.WriteLine($"消息 >>>>>： {data} \r\n發送到：{report.TopicPartitionOffset}");
}
catch (ProduceException<string, string> ex)
{
Console.WriteLine($"消息發送失敗>>>>>：\r\n Code= {ex.Error.Code} >>> \r\nError= {ex.Message}");
}
}
}
} View Code
36、新增的消費者接收服務方法代碼如下：
/// <summary>
/// Description: kafka 消費者訂閱服務
/// CreateTime: 2022/1/21 19:36:25
/// Author: Wesky
/// </summary>
public class SubscribeService: ISubscribeService
{
/// <summary>
/// 消費者服務核心代碼
/// </summary>
/// <typeparam name="TMessage"></typeparam>
/// <param name="config">消費者配置資訊</param>
/// <param name="topics">主題集合</param>
/// <param name="func"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public async Task SubscribeAsync<TMessage>(ConsumerConfig config, IEnumerable<string> topics, Action<TMessage> func, CancellationToken cancellationToken) where TMessage : class
{
const int commitPeriod = 1;
using (var consumer = new ConsumerBuilder<Ignore, string>(config)
.SetErrorHandler((_, e) =>
{
Console.WriteLine($"消費錯誤 >>>>>: {e.Reason}");
})
.SetStatisticsHandler((_, json) =>
{
Console.WriteLine($"************************************************");
})
.SetPartitionsAssignedHandler((c, partitionList) =>
{
string partitions = string.Join(", ", partitionList);
Console.WriteLine($"配置設定的分區 >>>>> : {partitions}");
})
.SetPartitionsRevokedHandler((c, partitionList) =>
{
string partitions = string.Join(", ", partitionList);
Console.WriteLine($"回收的分區 >>>>> : {partitions}");
})
.Build())
{
consumer.Subscribe(topics);
try
{
while (true)
{
try
{
var consumeResult = consumer.Consume(cancellationToken);
if (consumeResult.IsPartitionEOF)
{
continue;
}
if(consumeResult?.Offset % commitPeriod == 0){
try
{
var result = JsonConvert.DeserializeObject<TMessage>(consumeResult.Message?.Value);
func(result); // 消費消息
}
catch (Exception ex)
{
Console.WriteLine($"消費業務處理失敗: {ex.Message}");
}
try
{
consumer.Commit(consumeResult); // 手動送出
Console.WriteLine($"消費者消費完成，已送出 ");
}
catch (KafkaException e)
{
Console.WriteLine($"送出錯誤 >>>>> : {e.Error.Reason}");
}
}
}
catch (ConsumeException e)
{
Console.WriteLine($"消費錯誤>>>>> : {e.Error.Reason}");
}
}
}
catch (Exception e)
{
Console.WriteLine($"其他錯誤 >>>>> :{e.Message}");
consumer.Close();
}
}
await Task.CompletedTask;
}
} 37、并且提供對應的接口服務，用于開放給外部調用，或者提供依賴注入使用：
38、建立一個控制台項目，用來當作消費者端的測試，并且新增一個方法，用來當作消費者接收到消息以後的業務處理方法體。此處控制台環境版本為.NET 6
39、消費用戶端代碼如下。其中，BootstrapServers也可以提供叢集位址，例如 ip1:port,ip2:port…… 服務之間以半形逗號隔開。
40、再新增一個webapi項目，用來當作生産者的用戶端進行發送資料。以及對kafka服務類部分進行依賴注入注冊，此處使用單例。該webapi此處使用.NET 6環境，帶有控制器的模式。
41、新增的控制器裡面，進行生産者的注入與實作。注意：topicName參數對應上邊的topic-wesky,通過主題綁定，否則消費者不認識就沒辦法消費到了。
控制器代碼：
[Route("api/[controller]/[action]")]
[ApiController]
public class ProducerController : ControllerBase
{
IPublishService _service = null;
public ProducerController(IPublishService publishService)
{
_service = publishService;
}
[HttpPost]
public IActionResult SendMessage(string broker,string topicName,string message)
{
_service.PublishAsync(broker, topicName, message);
return Ok();
}
} 42、接下來是一些測試，如圖所示：
43、最後，使用可視化管理工具Offset進行檢視，可以看到對應的主題。選中主題，可以設定資料類型，這裡我設定為字元串，就可以檢視到對應的消息内容了。如果沒有設定，預設是16進制的資料。
44、檢視剛剛測試時候收發的消息隊列裡面的資料，如下所示：
45、一些額外補充：
Kafka也是消息隊列的一種，用于在高吞吐量場景下使用比較适合。如果是輕量級的，隻需要用于削峰，可以使用RabbitMQ。
以上隻是簡單的操作示範，至于要用得溜，觀衆朋友們可以自行補充所需的相關理論知識。
可視化工具還有一款yahoo提供的開源的工具，叫kafka-manager，有興趣的大佬們可以自行玩玩，開源位址：
https://github.com/yahoo/CMAK
還有一款滴滴平台做的開源的kafka運維管理平台，有興趣的大佬們也可以自行了解，位址：
https://github.com/didi/LogiKM
以上就是該部落格的全部内容，感謝各位大佬們的觀看~
歡迎加入QQ群：
群号：1079830632