kafka学习 （三）csharp开发

1 kafka-net

1.1全局配置

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
<startup>
<supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.6.1" />
</startup>
<appSettings>
<add key="KafkaBroker" value="http://26.2.4.171:9092,http://26.2.4.172:9092,http://26.2.4.173:9092" />
<add key="Topic" value="test_topic" />
</appSettings>
</configuration>

1.2 producer

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using KafkaNet;
using KafkaNet.Common;
using KafkaNet.Model;
using KafkaNet.Protocol;
using System.Configuration;

namespace producer
{
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)
{

string[] temp = broker.Split(',');
Uri[] url = new Uri[3];
int index = 0;
foreach (string item in temp)
{
url[index] = new Uri(item);
index++;
}
var options = new KafkaOptions(url);
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(2000);
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;
}
}
}

1.3 consumer

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using KafkaNet;
using KafkaNet.Common;
using KafkaNet.Model;
using KafkaNet.Protocol;
using System.Configuration;

namespace comsumer
{
class Program
{
static void Main(string[] args)
{
Consume(getKafkaBroker(), getTopicName());

}

private static void Consume(string broker, string topic)
{
string[] temp = broker.Split(',');
Uri[] url =new Uri[3];
int index = 0;
foreach(string item in temp)
{
url[index] = new Uri(item);
index++;

15443
}
var options = new KafkaOptions(url);
var router = new BrokerRouter(options);
OffsetPosition[] off =new OffsetPosition[3];
//off
off[0] = new OffsetPosition(0, 9999);
off[1] = new OffsetPosition(1, 9999);
off[2] = new OffsetPosition(2, 9999);
Consumer consumer = new Consumer(new ConsumerOptions(topic, router),off);
//var tt2 = consumer.GetTopicOffsetAsync(topic, 2, -1);
////var tt = consumer.GetOffsetPosition();
//var test = CreateOffsetFetchRequest(topic,0);
//var test2 = CreateFetchRequest(topic, 0);
////consumer.SetOffsetPosition(new OffsetPosition(consumer.GetOffsetPosition()));
////Consume returns a blocking IEnumerable (ie: never ending stream)
bool flag = false;
//var t =consumer.Consume();
//var tt = consumer.GetOffsetPosition();
List<OffsetPosition> tt3;
foreach (var message in consumer.Consume())
{
//if (!flag)
//{

tt3 = consumer.GetOffsetPosition();
OffsetPosition[] str = tt3.ToArray();
consumer.SetOffsetPosition(str);
//consumer.Dispose();
//break;

flag = true;
//}
Console.WriteLine("1 Response: Partition {0},Offset {1} : {2}",
message.Meta.PartitionId, message.Meta.Offset, message.Value.ToUtf8String());
}
//consumer.Dispose();
//OffsetPosition[] str = tt3.ToArray();
//var consumer2 = new Consumer(new ConsumerOptions(topic, router), str);
//foreach (var message in consumer2.Consume())
//{

//    Console.WriteLine("2 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;
}
}
}

2 rdkafka

zlib.dll
librdkafka.dll

2.1 producer

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using RdKafka;
namespace RDkafka_producer
{
public class Program
{
public static void Main(string[] args)
{
string brokerList = "26.2.4.171:9092,26.2.4.172:9092,26.2.4.173:9092";
string topicName = "test_topic";

var topicConfig = new TopicConfig
{
//CustomPartitioner = (top, key, cnt) =>
//{
//    var kt = (key != null) ? Encoding.UTF8.GetString(key, 0, key.Length) : "(null)";

//    int partition = 0;
//    if (key != null)
//    {
//        partition = key.Length % cnt;
//    }

//    bool available = top.PartitionAvailable(partition);
//    Console.WriteLine("Partitioner topic: {"+top.Name+"} key: {"+kt+"} partition count: {"+cnt+"} -> {"+partition+"} {"+available+"}");
//    return partition;
//}
};

using (Producer producer = new Producer(brokerList))
using (Topic topic = producer.Topic(topicName, topicConfig))
{
Console.WriteLine("{"+producer.Name+"} producing on {"+topic.Name+"}. q to exit.");

string text;
while ((text = Console.ReadLine()) != "q")
{
byte[] data = Encoding.UTF8.GetBytes(text);
Task<DeliveryReport> deliveryReport = topic.Produce(data);
var unused = deliveryReport.ContinueWith(task =>
{
Console.WriteLine("Partition: {"+task.Result.Partition+"}, Offset: {"+task.Result.Offset+"}");
});
}
}
}
}
}

2.2 consumer

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using RdKafka;

namespace RD_kafka_consumer
{
public class Program
{
public static void Run(string brokerList, List<string> topics)
{
bool enableAutoCommit = false;

var config = new Config()
{
GroupId = "advanced-csharp-consumer",
EnableAutoCommit = enableAutoCommit,
StatisticsInterval = TimeSpan.FromSeconds(60)
};

using (var consumer = new EventConsumer(config, brokerList))
{
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}");
Console.WriteLine("1 Response: Partition {0},Offset {1} : {2}",
msg.Partition, msg.Offset, text);
if (!enableAutoCommit && msg.Offset % 10 == 0)
{
Console.WriteLine("Committing offset");
consumer.Commit(msg).Wait();
Console.WriteLine("Committed offset");
}
};

consumer.OnConsumerError += (obj, errorCode) =>
{
//Console.WriteLine($"Consumer Error: {errorCode}");
};

consumer.OnEndReached += (obj, end) => {
//Console.WriteLine($"Reached end of topic {end.Topic} partition {end.Partition}, next message will be at offset {end.Offset}");
};

consumer.OnError += (obj, error) => {
//Console.WriteLine($"Error: {error.ErrorCode} {error.Reason}");
};

if (enableAutoCommit)
{
consumer.OnOffsetCommit += (obj, commit) => {
if (commit.Error != ErrorCode.NO_ERROR)
{
//Console.WriteLine($"Failed to commit offsets: {commit.Error}");
}
//Console.WriteLine($"Successfully committed offsets: [{string.Join(", ", commit.Offsets)}]");
};
}

consumer.OnPartitionsAssigned += (obj, partitions) => {
//Console.WriteLine($"Assigned partitions: [{string.Join(", ", partitions)}], member id: {consumer.MemberId}");
consumer.Assign(partitions);
};

consumer.OnPartitionsRevoked += (obj, partitions) => {
//Console.WriteLine($"Revoked partitions: [{string.Join(", ", partitions)}]");
consumer.Unassign();
};

consumer.OnStatistics += (obj, json) => {
//Console.WriteLine($"Statistics: {json}");
};

consumer.Subscribe(topics);
consumer.Start();

//Console.WriteLine($"Assigned to: [{string.Join(", ", consumer.Assignment)}]");
//Console.WriteLine($"Subscribed to: [{string.Join(", ", consumer.Subscription)}]");

//Console.WriteLine($"Started consumer, press enter to stop consuming");
Console.ReadLine();
}
}
public static void Main(string[] args)
{
string brokerlist = "26.2.4.171:9092,26.2.4.172:9092,26.2.4.173:9092";
List<string> topic = new List<string>();
topic.Add("test_topic");
Run(brokerlist, topic);
}
}
}

2.3 参数配置

rd_kafka_brokers_add()

metadata.broker.list

max.in.flight.requests.per.connection

rd_kafka_conf_set_stats_cb()

rd_kafka_conf_set_events()

connection.max.idle.ms

broker.version.fallback

broker.version.fallback

broker.version.fallback

api.version.request

api.version.fallback.ms

ciphers(1)

... -t %{sasl.kerberos.keytab}

fetch.message.max.bytes

message.send.max.retries

compression.codec

Topic configuration properties

in.sync.replicas

request.required.acks

request.required.acks

dr_msg_cb

rd_kafka_message_t.offset

rd_kafka_offset_store()

enable.auto.commit

auto.commit.enable

rd_kafka_consume_callback*()