新建一个cmd窗口，zkServer命令启动zookeeper
打开另一个cmd窗口,输入:
cd D:\Worksoftware\Apachekafka2.11\bin\windows
kafka-server-start D:\Worksoftware\Apachekafka2.11\config\server.properties
删除主题：E:\WorkSoftWare\kafka2.11\bin\windows>kafka-run-class.bat kafka.admin.TopicCommand --delete --topic TestSiso --zookeeper localhost:2181
kafka 删除topic 提示marked for deletion
并没有真正删除，如果要真正删除
在每一台机器中的kafka_2.10/config/server.properties 文件加入 delete.topic.enable=true
最后所有机器重新启动kafka
启动kafka成功后，就可以运行项目了
引用了kafka-net.dll
Program.cs
internal class Program
{
private static void Main(string[] args)
{
string header = "kafka测试";
Console.Title = header;
Console.WriteLine(header);
ConsoleColor color = Console.ForegroundColor;
var pub = new KafkaHelper("Test", true);
var sub = new KafkaHelper("Test", false);
Task.Run(() =>
{
while (true)
{
string msg = string.Format("{0}这是一条测试消息", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff"));
pub.Pub(new List<string> {msg});
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine("发送消息：" + msg);
//Console.ForegroundColor = color;
Thread.Sleep(2000);
}
});
Task.Run(() => sub.Sub(msg =>
{
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("收到消息：{0}", msg);
//Console.ForegroundColor = color;
}));
Console.ReadLine();
}
}
KafkaHelper.cs代码:
/// <summary>
/// kafka辅助类
/// </summary>
public sealed class KafkaHelper
{
private readonly KafkaConfig _config;
private readonly ConsumerHelper _consumerHelper;
private readonly bool _isProducer = true;
private readonly ProduceHelper _produceHelper;
private BrokerHelper _brokerHelper;
/// <summary>
/// kafka辅助类构造方法
/// </summary>
/// <param name="sectionName">config中配置节点名称</param>
/// <param name="isProducer"></param>
public KafkaHelper(string sectionName, bool isProducer = true)
{
_isProducer = isProducer;
_config = KafkaConfig.GetConfig(sectionName);
_brokerHelper = new BrokerHelper(_config.Broker);
if (isProducer)
_produceHelper = new ProduceHelper(_brokerHelper);
else
_consumerHelper = new ConsumerHelper(_brokerHelper);
}
/// <summary>
/// 是否是生产者模式
/// </summary>
public bool IsProducer
{
get { return _isProducer; }
}
/// <summary>
/// 发送消息到队列
/// </summary>
/// <param name="topic"></param>
/// <param name="datas"></param>
/// <param name="acks"></param>
/// <param name="timeout"></param>
public void Pub(List<string> datas, short acks = 1, TimeSpan? timeout = default(TimeSpan?))
{
_produceHelper.Pub(_config.Topic, datas, acks, timeout, MessageCodec.CodecNone);
}
/// <summary>
/// 订阅消息
/// </summary>
/// <param name="onMsg"></param>
public void Sub(Action<string> onMsg)
{
_consumerHelper.Sub(_config.Topic, onMsg);
}
/// <summary>
/// 取消订阅
/// </summary>
public void UnSub()
{
_consumerHelper.UnSub();
}
}
KafkaConfig.cs代码:
/// <summary>
/// kafka配置类
/// </summary>
public class KafkaConfig : ConfigurationSection
{
/// <summary>
/// 当前配置名称
/// 此属性为必须
/// </summary>
public string SectionName { get; set; }
/// <summary>
/// 代理
/// </summary>
[ConfigurationProperty("broker", IsRequired = true)]
public string Broker
{
get { return (string) base["broker"]; }
set { base["broker"] = value; }
}
/// <summary>
/// 主题
/// </summary>
[ConfigurationProperty("topic", IsRequired = true)]
public string Topic
{
get { return (string) base["topic"]; }
set { base["topic"] = value; }
}
#region 从配置文件中创建kafka配置类
/// <summary>
/// 获取默认kafka配置类
/// </summary>
/// <returns></returns>
public static KafkaConfig GetConfig()
{
return (KafkaConfig) ConfigurationManager.GetSection("kafkaConfig");
}
/// <summary>
/// 获取指定的kafka配置类
/// </summary>
/// <param name="sectionName"></param>
/// <returns></returns>
public static KafkaConfig GetConfig(string sectionName)
{
var section = (KafkaConfig) ConfigurationManager.GetSection(sectionName);
// 跟默认配置相同的，可以省略
if (section == null)
section = GetConfig();
if (section == null)
throw new ConfigurationErrorsException("kafkacofng节点 " + sectionName + " 未配置.");
section.SectionName = sectionName;
return section;
}
/// <summary>
/// 从指定位置读取配置
/// </summary>
/// <param name="fileName"></param>
/// <param name="sectionName"></param>
/// <returns></returns>
public static KafkaConfig GetConfig(string fileName, string sectionName)
{
return GetConfig(ConfigurationManager.OpenMappedMachineConfiguration(new ConfigurationFileMap(fileName)),
sectionName);
}
/// <summary>
/// 从指定Configuration中读取配置
/// </summary>
/// <param name="config"></param>
/// <param name="sectionName"></param>
/// <returns></returns>
public static KafkaConfig GetConfig(Configuration config, string sectionName)
{
if (config == null)
throw new ConfigurationErrorsException("传入的配置不能为空");
var section = (KafkaConfig) config.GetSection(sectionName);
if (section == null)
throw new ConfigurationErrorsException("kafkacofng节点 " + sectionName + " 未配置.");
section.SectionName = sectionName;
return section;
}
#endregion
}
BrokerHelper.cs代码:
/// <summary>
/// 代理人辅助类
/// </summary>
internal class BrokerHelper
{
private readonly string _broker;
public BrokerHelper(string broker)
{
_broker = broker;
}
/// <summary>
/// 获取代理的路由对象
/// </summary>
/// <returns></returns>
public BrokerRouter GetBroker()
{
var options = new KafkaOptions(new Uri(string.Format("http://{0}", _broker)));
return new BrokerRouter(options);
}
}
ConsumerHelper.cs代码:
/// <summary>
/// 消费者辅助类
/// </summary>
internal class ConsumerHelper
{
private readonly BrokerHelper _brokerHelper;
private Consumer _consumer;
private bool _unSub;
public ConsumerHelper(BrokerHelper brokerHelper)
{
_brokerHelper = brokerHelper;
}
public void Sub(string topic, Action<string> onMsg)
{
_unSub = false;
var opiton = new ConsumerOptions(topic, _brokerHelper.GetBroker());
_consumer = new Consumer(opiton);
Task.Run(() =>
{
while (!_unSub)
{
IEnumerable<Message> msgs = _consumer.Consume();
Parallel.ForEach(msgs, msg => onMsg(Encoding.UTF8.GetString(msg.Value)));
}
});
}
public void UnSub()
{
_unSub = true;
}
}
ProduceHelper.cs代码:
/// <summary>
/// 生产者辅助类
/// </summary>
internal class ProduceHelper : IDisposable
{
private readonly Producer _producer;
private BrokerHelper _brokerHelper;
public ProduceHelper(BrokerHelper brokerHelper)
{
_brokerHelper = brokerHelper;
_producer = new Producer(_brokerHelper.GetBroker());
}
public void Dispose()
{
if (_producer != null)
_producer.Dispose();
}
/// <summary>
/// 发送消息到队列
/// </summary>
/// <param name="topic"></param>
/// <param name="datas"></param>
/// <param name="acks"></param>
/// <param name="timeout"></param>
/// <param name="codec"></param>
public void Pub(string topic, List<string> datas, short acks = 1, TimeSpan? timeout = default(TimeSpan?), MessageCodec codec = MessageCodec.CodecNone)
{
//var msgs = new List<Message>();
//foreach (string item in datas)
//{
// msgs.Add(new Message(item));
//}
var msgs = datas.Select(item => new Message(item)).ToList();
_producer.SendMessageAsync(topic, msgs, acks, timeout, codec);
}
}
App.config
<?xml version="1.0" encoding="utf-8"?>
<configuration>
<configSections>
<section name="Test" type="xxxxx.sssss.KafkaConfig, xxxxx.sssss" />
</configSections>
<Test broker="127.0.0.1:9092" topic="TestSiso" />
<startup>
<supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.5" />
</startup>
</configuration>