本文介绍如何使用C# SDK通过接入点接入消息队列Kafka版并收发消息。
环境配置
您已安装.NET。更多信息,请参见安装.NET。
安装C#依赖库
执行以下命令安装C#依赖库。
dotnet add package -v 1.5.2 Confluent.Kafka
准备配置
发送消息
执行以下命令发送消息。
dotnet run producer.cs
消息程序producer.cs示例代码如下:
关于代码中配置项说明,请参见配置项说明。
注意 示例代码为SSL接入点的代码。您需要根据实际接入点类型,删除或者修改配置项代码。
using System;
using Confluent.Kafka;
class Producer
{
public static void Main(string[] args)
{
var conf = new ProducerConfig {
BootstrapServers = "XXX,XXX,XXX",
SslCaLocation = "XXX/only-4096-ca-cert.pem",
SaslMechanism = SaslMechanism.Plain,
SecurityProtocol = SecurityProtocol.SaslSsl,
SaslUsername = "XXX",
SaslPassword = "XXX",
};
Action<DeliveryReport<Null, string>> handler = r =>
Console.WriteLine(!r.Error.IsError
? $"Delivered message to {r.TopicPartitionOffset}"
: $"Delivery Error: {r.Error.Reason}");
string topic ="XXX";
using (var p = new ProducerBuilder<Null, string>(conf).Build())
{
for (int i=0; i<100; ++i)
{
p.Produce(topic, new Message<Null, string> { Value = i.ToString() }, handler);
}
p.Flush(TimeSpan.FromSeconds(10));
}
}
}
订阅消息
执行以下命令消费消息。
dotnet run consumer.cs
消息程序consumer.cs示例代码如下:
关于代码中配置项说明,请参见配置项说明。
注意 示例代码为SSL接入点的代码。您需要根据实际接入点类型,删除或者修改配置项代码。
using System;
using System.Threading;
using Confluent.Kafka;
class Consumer
{
public static void Main(string[] args)
{
var conf = new ConsumerConfig {
GroupId = "XXX",
BootstrapServers = "XXX,XXX,XXX",
SslCaLocation = "XXX/only-4096-ca-cert.pem",
SaslMechanism = SaslMechanism.Plain,
SecurityProtocol = SecurityProtocol.SaslSsl,
SaslUsername = "XXX",
SaslPassword = "XXX",
AutoOffsetReset = AutoOffsetReset.Earliest
};
string topic = "XXX";
using (var c = new ConsumerBuilder<Ignore, string>(conf).Build())
{
c.Subscribe(topic);
CancellationTokenSource cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) => {
e.Cancel = true;
cts.Cancel();
};
try
{
while (true)
{
try
{
var cr = c.Consume(cts.Token);
Console.WriteLine($"Consumed message '{cr.Value}' at: '{cr.TopicPartitionOffset}'.");
}
catch (ConsumeException e)
{
Console.WriteLine($"Error occured: {e.Error.Reason}");
}
}
}
catch (OperationCanceledException)
{
c.Close();
}
}
}
}