windowskafka_0">在windows系统中安装kafka配置全步骤记录

提示：这里可以添加系列文章的所有文章的目录，目录需要自己手动添加
例如：第一章 Python 机器学习入门之pandas的使用

---

提示：写完文章后，目录可以自动生成，如何生成可参考右边的帮助文档

---

前言

提示：这里可以添加本文要记录的大概内容：

例如：随着人工智能的不断发展，机器学习这门技术也越来越重要，很多人都开启了学习机器学习，本文就介绍了机器学习的基础内容。

---

提示：以下是本篇文章正文内容，下面案例可供参考

一、Java JDK安装

1.1 官网下载地址 JDK下载

1.2 配置环境变量（网上一大堆，随便搜）。

二、Zookeeper安装

Kafka是因为kafka是基于Zookeeper的，而Zookeeper一般都是一个分布式的集群，尽管kafka有自带Zookeeper，但是一般不使用自带的，都是使用外部安装的，所以首先我们需要安装Zookeeper。

2.1 官网下载地址 Zookeeper官网

2.2 本地安装

	1.把下载好的压缩包解压，放到自己指定想要安装的目录下；2.在bin文件夹同级下，创建data文件夹、log文件夹;3.找到config文件夹，将目录中zoo_sample.cfg文件复制一份，重命名为zoo.cfg;4.修改zoo.cfg配置文件，将默认的 dataDir=/tmp/zookeeper 修改成 zookeeper 安装目录所在的data 文件夹，再增加数据日志的配置；

2.3 启动测试

	完成上面所有配置后，进入到zookeeper安装目录下的bin文件下，双击运行 zkServer.cmd 启动；保持cmd窗口不要关掉！！！

三、Kafka安装

1.官网下载 kafka.apache.org/downloads.html" rel="nofollow">kafka官网

2.修改文件夹名字（名字不宜过长）

3.修改配置文件中

	解压到相应文件夹，进入config目录下，找到server.properties文件

kafkacmdkafka_binwindows_46">4.启动kafka，cmd到kafka解压文件的 .\bin\windows\文件夹下,执行

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

四、offset安装

offset是kafka的可视化工具，方便查看，下载地址 [offset下载地址](https://www.kafkatool.com/download2/offsetexplorer_64bit.exe)

五、生产者代码

首先要保证安装 Confluent.Kafka

using Confluent.Kafka;
using System;
using System.Threading.Tasks;namespace KafkaProducer
{class Program{ static void Main(string[] args){Console.WriteLine("请输入消息内容");using (var producer = new KafkaProducer()){while (true){string message = Console.ReadLine();try{//topic名称是testvar result = producer.ProduceAsync("topic01",new Confluent.Kafka.Message<string, string>() { Key = Guid.NewGuid().ToString(), Value = message }).GetAwaiter().GetResult();Console.WriteLine($"offset:{result.Offset.Value},partition:{result.Partition.Value}");}catch (ProduceException<string, string> e){Console.WriteLine($"失败的消息: {e.Message} [{e.Error.Code}]");continue;}}}}class KafkaProducer : IDisposable{private ProducerConfig _config = new ProducerConfig();private IProducer<string, string> _producer;public KafkaProducer(string server = null){if (string.IsNullOrEmpty(server)){//这里可以添加更多的Kafka集群，比如//server=" server ="192.168.1.129:9092,192.168.1.133:9092,192.168.1.134:9092";";                   server = "192.168.250.11:9092";}_config.BootstrapServers = server;_producer = new ProducerBuilder<string, string>(_config).Build();}public async Task<DeliveryResult<string, string>> ProduceAsync(string topic, Message<string, string> message){return await _producer.ProduceAsync(topic, message);}public void Dispose(){_producer?.Dispose();}}}
}

六、消费者代码

using Confluent.Kafka;
using System;namespace KafkaConsumer
{class Program{static void Main(string[] args){Console.WriteLine("默认只关注test主题的消息)");using (var consumer = new KafkaConsumer()){while (true){consumer.Consume(a =>{if (a == null){Console.WriteLine("暂无消息");}else{Console.WriteLine($"Key:{a.Message.Key},Value:{a.Message.Value}");}});}}}class KafkaConsumer : IDisposable{private IConsumer<string, string> _consumer;public KafkaConsumer(string server = null){if (string.IsNullOrEmpty(server)){server = "192.168.250.11:9092";}var config = new ConsumerConfig{GroupId = "group.1",BootstrapServers = server,AutoOffsetReset = AutoOffsetReset.Earliest};_consumer = new ConsumerBuilder<string, string>(config).Build();//topic名称默认是test_consumer.Subscribe("topic01");}public void Consume(Action<ConsumeResult<string, string>> action = null){var consumerResult = _consumer.Consume(TimeSpan.FromSeconds(1));action?.Invoke(consumerResult);}public void Dispose(){_consumer?.Dispose();}}}
}}}}
}