Win10安装kafka并用C#调用

 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),如下代码举行消息发布和订阅。

1.        private async void simpleButton2_Click(object sender, EventArgs e)
2.        {
3.            var config = new ProducerConfig
4.            {
5.                BootstrapServers = "localhost:9092"
6.            };
7.            using (var producer = new ProducerBuilder<Null, string>(config).Build())
8.            {
9.                var dr = await producer.ProduceAsync("test-topic", new Message<Null, string> { Value = "hello,world" });
10.                Debug.WriteLine($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");
11.            }
13.        }
15.        private void simpleButton3_Click(object sender, EventArgs e)
16.        {
17.            Task.Factory.StartNew(() => {
18.                var config = new ConsumerConfig
19.                {
20.                    GroupId = "test-consumer-group",
21.                    BootstrapServers = "localhost:9092",
22.                    AutoOffsetReset = AutoOffsetReset.Earliest
23.                };
24.                var cts = new CancellationTokenSource();
25.                using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
26.                {
27.                    consumer.Subscribe("test-topic");
28.                   
29.                    try
30.                    {
31.                        while (true)
32.                        {
33.                            try
34.                            {
35.                                var cr = consumer.Consume(cts.Token);
36.                                Debug.WriteLine($"Consumed message '{cr.Value}' at: '{cr.TopicPartitionOffset}'.");
37.                            }
38.                            catch (Exception ex)
39.                            {
40.                                Console.WriteLine(ex);
41.                            }
42.                        }
43.                    }
44.                    catch (Exception ex2)
45.                    {
46.                        Console.WriteLine(ex2);
47.                    }
48.                }
49.            });
51.        }

复制代码

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