赣榆网站建设国家税务总局网

 kafka安装

jdk、kafka版本如下，zookeeper使用kafka自带版本

安装包下载位置:https://download.csdn.net/download/henreash/90087368 (赚点csdn下载资源分)

安装jdk后，解压kafka压缩包，修改配置文件：

kafka_2.13-3.9.0\config\zookeeper.properties内修改：dataDir=D:/Kafka/zookeeper/data

kafka_2.13-3.9.0\config\server.properties内修改：log.dirs=D:/Kafka/kafka-logs

在目录内创建批处理文件，启动zookeeper和kafka：

kfk.cmd内容：call bin/windows/kafka-server-start.bat config/server.properties

zk.bat内容：call bin/windows/zookeeper-server-start.bat config/zookeeper.properties

双击zk.bat、kfk.cmd启动zookeeper和kafka

kafka-manager安装

将kafka-manager2解压到d盘，注意目录结构不能太深，否则启动报错。

修改config\application.conf，修改节点kafka-manager.zkhosts指向zookeeper

 kafka-manager.zkhosts="192.168.0.109:2181"

创建start.bat批处理文件，内容：.\bin\kafka-manager.bat

双击启动kafka-manager。

打开浏览器，输入地址：http://localhost:9000/

点击Cluster菜单，创建一个默认Cluster Test001；

kafka环境配置完毕。

C#调用

创建C#8项目，在nuget中下载Confluent.Kafka(2.6.1),如下代码进行消息发布和订阅。

       private async void simpleButton2_Click(object sender, EventArgs e){var config = new ProducerConfig { BootstrapServers = "localhost:9092"};using (var producer = new ProducerBuilder<Null, string>(config).Build()){var dr = await producer.ProduceAsync("test-topic", new Message<Null, string> { Value = "hello,world" });Debug.WriteLine($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");}}private void simpleButton3_Click(object sender, EventArgs e){Task.Factory.StartNew(() => {var config = new ConsumerConfig{GroupId = "test-consumer-group",BootstrapServers = "localhost:9092",AutoOffsetReset = AutoOffsetReset.Earliest};var cts = new CancellationTokenSource();using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build()){consumer.Subscribe("test-topic");try{while (true){try{var cr = consumer.Consume(cts.Token);Debug.WriteLine($"Consumed message '{cr.Value}' at: '{cr.TopicPartitionOffset}'.");}catch (Exception ex){Console.WriteLine(ex);}}}catch (Exception ex2){Console.WriteLine(ex2);}}});}

执行后kafka-manager界面如下图：