.net +RdKafka

先在nuget里安装RdKafka

using RdKafka;
using System;
using System.Text;

namespace KafkaApp
{
class Program
{
static void Main(string[] args)
{

//ProducterTest();
ConsumerTest();
}
private static async void ProducterTest()
{
// Producer 接受一个或多个 BrokerList
using (Producer producer = new Producer("192.168.1.4:9092"))
//发送到一个名为 testtopic 的Topic，如果没有就会创建一个
using (Topic topic = producer.Topic("666"))
{
//将message转为一个 byte[]
byte[] data = Encoding.UTF8.GetBytes("Hello RdKafka");
DeliveryReport deliveryReport = await topic.Produce(data);

Console.WriteLine($"发送到分区：{deliveryReport.Partition}, Offset 为: {deliveryReport.Offset}");
}
}

public static void ConsumerTest()
{

//配置消费者组
var config = new Config() { GroupId = "example-csharp-consumer" };
using (var consumer = new EventConsumer(config, "192.168.1.4:9092"))
{

//注册一个事件
consumer.OnMessage += (obj, msg) =>
{
string text = Encoding.UTF8.GetString(msg.Payload, 0, msg.Payload.Length);
Console.WriteLine($"Topic: {msg.Topic} Partition: {msg.Partition} Offset: {msg.Offset} {text}");
};

//订阅一个或者多个Topic
consumer.Subscribe(new[] { "666" });

//启动
consumer.Start();

Console.WriteLine("Started consumer, press enter to stop consuming");
Console.ReadLine();
}
}
}
}