首页 > 代码库 > net windows Kafka
net windows Kafka
net windows Kafka 安装与使用入门(入门笔记)
完整解决方案请参考：
Setting Up and Running Apache Kafka on Windows OS
在环境搭建过程中遇到两个问题，在这里先列出来，以方便查询：
1. \Java\jre7\lib\ext\QTJava.zip was unexpected at this time. Process exited
解决方案：
1.1 右键点击“我的电脑” -> "高级系统设置" -> "环境变量"
1.2 查看CLASSPATH的值里有没有包含QTJava.zip的路径，如果有的话，把对应的路径删除，问题就解决了。
2. Missing `server‘ JVM (Java\jre7\bin\server\jvm--.dll.)
解决方案：
2.1. 拷贝C:\Program Files\Java\jdk1.6.0\jre\bin\server
2.2 粘贴到 C:\Program Files\Java\jre1.6.0\bin
搭建环境
1. 安装JDK
1.1 安装文件：http://www.oracle.com/technetwork/java/javase/downloads/jre8-downloads-2133155.html下载Server JRE.
1.2 安装完成后需要添加以下的环境变量（右键点击“我的电脑” -> "高级系统设置" -> "环境变量" ）：
-
- JAVA_HOME: C:\Program Files (x86)\Java\jre1.8.0_60（这个是默认安装路径，如果安装过程中更改了安装目录，把更改后的路径填上就行了）
-
- PATH: 在现有的值后面添加"; %JAVA_HOME%\bin"
1.3 打开cmd运行 "java -version" 查看当前系统Java的版本：
2. 安装Zookeeper
Kafka的运行依赖于Zookeeper，所以在运行Kafka之前我们需要安装并运行Zookeeper
2.1 下载安装文件： http://zookeeper.apache.org/releases.html
2.2 解压文件（本文解压到 G:\zookeeper-3.4.8）
2.3 打开G:\zookeeper-3.4.8\conf，把zoo_sample.cfg重命名成zoo.cfg
2.4 从文本编辑器里打开zoo.cfg
2.5 把dataDir的值改成“:\zookeeper-3.4.8\data”
2.6 添加如下系统变量：
-
- ZOOKEEPER_HOME: G:\zookeeper-3.4.8
- Path: 在现有的值后面添加 ";%ZOOKEEPER_HOME%\bin;"
2.7 运行Zookeeper: 打开cmd然后执行
zkserver
3. 安装并运行Kafka
3.1 下载安装文件： http://kafka.apache.org/downloads.html
3.2 解压文件（本文解压到 G:\kafka_2.11-0.10.0.1）
3.3 打开G:\kafka_2.11-0.10.0.1\config
3.4 从文本编辑器里打开 server.properties
3.5 把 log.dirs的值改成 “G:\kafka_2.11-0.10.0.1\kafka-logs”
3.6 打开cmd
3.7 进入kafka文件目录: cd /d G:\kafka_2.11-0.10.0.1\
3.8 输入并执行以打开kafka:
.\bin\windows\kafka-server-start.bat .\config\server.properties
4. 创建topics
4.1 打开cmd 并进入G:\kafka_2.11-0.10.0.1\bin\windows
4.2 创建一个topic：
kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
5. 打开一个Producer：
cd /d G:\kafka_2.11-0.10.0.1\bin\windows kafka-console-producer.bat --broker-list localhost:9092 --topic test
6. 打开一个Consumer:
cd /d G:\kafka_2.11-0.10.0.1\bin\windows
kafka-console-consumer.bat --zookeeper localhost:2181 --topic test
然后就可以在Producer控制台窗口输入消息了。在消息输入过后，很快Consumer窗口就会显示出Producer发送的消息：
至此，Kafka运行环境的搭建就完成了:-)
web管理方面（正式运维部署）：
网上有人说 kafka manage
.net sdk方面：
有人建议:https://github.com/Microsoft/CSharpClient-for-Kafka （微软提供的sdk，看了源码个人感觉有些复杂）
还有
kafka-net 这个感觉会简单很多（看了源码感觉一些Task写法并不喜欢）
推荐:kafka-net 简单易用些。
地址: https://github.com/Jroland/kafka-net
以上sdk 可以搜索nuget 查看开源地址和最新release包,当然.net sdk 还有很多其他选择。
个人开发工具（windows支持），windows 开发环境使用
kafkaTool
下载地址: http://www.kafkatool.com/download.html
kafka-net sdk使用demo
参考来源:http://www.cnblogs.com/Wulex/p/5619425.html
详细请查看官方开源地址的demo: https://github.com/Jroland/kafka-net
KafkaProducer程序:
class Program
{
static void Main(string[] args)
{
do
{
Produce(GetKafkaBroker(), getTopicName());
System.Threading.Thread.Sleep(3000);
} while (true);
}
private static void Produce(string broker, string topic)
{
var options = new KafkaOptions(new Uri(broker));
var router = new BrokerRouter(options);
var client = new Producer(router);
var currentDatetime =DateTime.Now;
var key = currentDatetime.Second.ToString();
var events = new[] { new Message("Hello World " + currentDatetime, key) };
client.SendMessageAsync(topic, events).Wait(1500);
Console.WriteLine("Produced: Key: {0}. Message: {1}", key, events[0].Value.ToUtf8String());
using (client) { }
}
private static string GetKafkaBroker()
{
string KafkaBroker = string.Empty;
const string kafkaBrokerKeyName = "KafkaBroker";
if (!ConfigurationManager.AppSettings.AllKeys.Contains(kafkaBrokerKeyName))
{
KafkaBroker = "http://localhost:9092";
}
else
{
KafkaBroker = ConfigurationManager.AppSettings[kafkaBrokerKeyName];
}
return KafkaBroker;
}
private static string getTopicName()
{
string TopicName = string.Empty;
const string topicNameKeyName = "Topic";
if (!ConfigurationManager.AppSettings.AllKeys.Contains(topicNameKeyName))
{
throw new Exception("Key \"" + topicNameKeyName + "\" not found in Config file -> configuration/AppSettings");
}
else
{
TopicName = ConfigurationManager.AppSettings[topicNameKeyName];
}
return TopicName;
}
}
4.KafkaConsumer程序:
class Program
{
static void Main(string[] args)
{
Consume(getKafkaBroker(), getTopicName());
}
private static void Consume(string broker, string topic)
{
var options = new KafkaOptions(new Uri(broker));
var router = new BrokerRouter(options);
var consumer = new Consumer(new ConsumerOptions(topic, router));
//Consume returns a blocking IEnumerable (ie: never ending stream)
foreach (var message in consumer.Consume())
{
Console.WriteLine("Response: Partition {0},Offset {1} : {2}",
message.Meta.PartitionId, message.Meta.Offset, message.Value.ToUtf8String());
}
}
private static string getKafkaBroker()
{
string KafkaBroker = string.Empty;
var KafkaBrokerKeyName = "KafkaBroker";
if (!ConfigurationManager.AppSettings.AllKeys.Contains(KafkaBrokerKeyName))
{
KafkaBroker = "http://localhost:9092";
}
else
{
KafkaBroker = ConfigurationManager.AppSettings[KafkaBrokerKeyName];
}
return KafkaBroker;
}
private static string getTopicName()
{
string TopicName = string.Empty;
var TopicNameKeyName = "Topic";
if (!ConfigurationManager.AppSettings.AllKeys.Contains(TopicNameKeyName))
{
throw new Exception("Key \"" + TopicNameKeyName + "\" not found in Config file -> configuration/AppSettings");
}
else
{
TopicName = ConfigurationManager.AppSettings[TopicNameKeyName];
}
return TopicName;
}
}
5.Consumer结果:
net windows Kafka
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。