接上篇Kafka的安装，我安装的Kafka集群地址：192.168.209.133:9092,192.168.209.134:9092,192.168.209.135:9092，所以这里直接使用这个集群来演示
首先创建一个项目，演示采用的是控制台（.net core 3.1），然后使用Nuget安装 Confluent.Kafka 包：
上面的截图中有Confluent.Kafka的源码地址，感兴趣的可以去看看：https://github.com/confluentinc/confluent-kafka-dotnet/
消息发布
先直接上Demo
static void Main(string[] args) { ProducerConfig config = new ProducerConfig(); config.BootstrapServers = "192.168.209.133:9092,192.168.209.134:9092,192.168.209.135:9092"; var builder = new ProducerBuilder<string, object>(config); builder.SetValueSerializer(new KafkaConverter());//设置序列化方式 var producer = builder.Build(); producer.Produce("test", new Message<string, object>() { Key = "Test", Value = "hello world" });
Console.ReadKey(); }
上述代码执行后，就可以使用上一节提到的kafkatool工具查看到消息了。
1、消息发布需要使用生产者对象，它由ProducerBuilder<,>类构造，有两个泛型参数，第一个是路由Key的类型，第二个是消息的类型，开发过程中，我们多数使用ProducerBuilder<string, object>或者ProducerBuilder<string, string>。
2、ProducerBuilder<string, object>在实例化时需要一个配置参数，这个配置参数是一个集合（IEnumerable<KeyValuePair<string, string>>），ProducerConfig其实是实现了这个集合接口的一个类型，在旧版本的Confluent.Kafka中，是没有这个ProducerConfig类型的，之前都是使用Dictionary<string,string>来构建ProducerBuilder<string, object>，比如上面的Demo，其实也可以写成：
static void Main(string[] args) { Dictionary<string, string> config = new Dictionary<string, string>(); config["bootstrap.servers"]= "192.168.209.133:9092,192.168.209.134:9092,192.168.209.135:9092"; var builder = new ProducerBuilder<string, object>(config); builder.SetValueSerializer(new KafkaConverter());//设置序列化方式 var producer = builder.Build(); producer.Produce("test", new Message<string, object>() { Key = "Test", Value = "hello world" }); Console.ReadKey(); }
这两种方式是一样的效果，只是ProducerConfig对象最终也是生成一个KeyValuePair<string, string>集合，ProducerConfig中的属性都会有一个Key与它对应，比如上面的ProducerConfig的BootstrapServers属性最终会映射成bootstrap.servers，表示Kafka集群地址，多个地址之间使用逗号分隔。
其他配置信息可以参考官方配置文档：https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
3、Confluent.Kafka还要求提供一个实现了ISerializer<TValue>或者IAsyncSerializer<TValue>接口的序列化类型，比如上面的Demo中的KafkaConverter：
public class KafkaConverter : ISerializer<object> { /// <summary> /// 序列化数据成字节 /// </summary> /// <param name="data"></param> /// <param name="context"></param> /// <returns></returns> public byte[] Serialize(object data, SerializationContext context) { var json = JsonConvert.SerializeObject(data); return Encoding.UTF8.GetBytes(json); } }
这里我采用的是Json格式序列化，需要使用Nuget安装Newtonsoft.Json。
4、发布消息使用Produce方法，它有几个重载，还有几个异步发布方法。第一个参数是topic，如果想指定Partition，需要使用TopicPartition对象，第二个参数是消息，它是Message<TKey, TValue>类型，Key即路由，Value就是我们的消息，消息会经过ISerializer<TValue>接口序列化之后发送到Kafka，第三个参数是Action<DeliveryReport<TKey, TValue>>类型的委托，它是异步执行的，其实是发布的结果通知。
消息消费
先直接上Demo
static void Main(string[] args) { ConsumerConfig config = new ConsumerConfig(); config.BootstrapServers = "192.168.209.133:9092,192.168.209.134:9092,192.168.209.135:9092"; config.GroupId = "group.1"; config.AutoOffsetReset = AutoOffsetReset.Earliest; config.EnableAutoCommit = false; var builder = new ConsumerBuilder<string, object>(config); builder.SetValueDeserializer(new KafkaConverter());//设置反序列化方式 var consumer = builder.Build(); consumer.Subscribe("test");//订阅消息使用Subscribe方法 //consumer.Assign(new TopicPartition("test", new Partition(1)));//从指定的Partition订阅消息使用Assign方法 while (true) { var result = consumer.Consume(); Console.WriteLine($"recieve message:{result.Message.Value}"); consumer.Commit(result);//手动提交，如果上面的EnableAutoCommit=true表示自动提交，则无需调用Commit方法 } }
1、和消息发布一样，消费者的构建是通过ConsumerBuilder<, >对象来完成的，同样也有一个ConsumerConfig配置对象，它在旧版本中也是不存在的，旧版本中也是使用Dictionary<string,string>来实现的，比如上面的例子等价于：
static void Main(string[] args) { Dictionary<string, string> config = new Dictionary<string, string>(); config["bootstrap.servers"] = "192.168.209.133:9092,192.168.209.134:9092,192.168.209.135:9092"; config["group.id"] = "group.1"; config["auto.offset.reset"] = "earliest"; config["enable.auto.commit"] = "false"; var builder = new ConsumerBuilder<string, object>(config); builder.SetValueDeserializer(new KafkaConverter());//设置反序列化方式 var consumer = builder.Build(); consumer.Subscribe("test");//订阅消息使用Subscribe方法 //consumer.Assign(new TopicPartition("test", new Partition(1)));//从指定的Partition订阅消息使用Assign方法 while (true) { var result = consumer.Consume(); Console.WriteLine($"recieve message:{result.Message.Value}"); consumer.Commit(result);//手动提交，如果上面的EnableAutoCommit=true表示自动提交，则无需调用Commit方法 } }
实际上，它和ProducerConfig一样也是一个KeyValuePair<string, string>集合，它的属性最终都会有一个Key与它对应。其他配置信息可以参考官方配置文档：https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
这里顺带提一下这个例子用到的几个配置：
BootstrapServers：Kafka集群地址，多个地址之间使用逗号分隔。
GroupId：消费者的Group，注意了，Kafka以Group的形式消费消息，一个消息只会被同一Group中的一个消费者消费，另外，一般的，同一Group中的消费者应该实现相同的逻辑
EnableAutoCommit：是否自动提交，如果设置成true，那么消费者接收到消息就相当于被消费了，我们可以设置成false，然后在我们处理完逻辑之后手动提交。
AutoOffsetReset：自动重置offset的行为，默认是Latest，这是kafka读取数据的策略，有三个可选值：Latest，Earliest，Error，个人推荐使用Earliest
关于AutoOffsetReset配置，这里再提一点
Latest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
Earliest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
Error：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
上面几句话说得有点蒙，举个例子：
当有一个消费者连接到Kafka，那这个消费者该从哪个位置开始消费呢？
首先，我们知道Kafka的消费者以群组Group的形式去消费，Kafka会记录每个Group在每个Partition中的到哪个位置，也就是offset。
当有消费者连接到Kafka要消费消息是，如果这个消费者所在的群组Group之前有消费过并提交过offset（也就是存在offset记录），那么这个消费者就从这个offset的位置开始消费，这一点Latest，Earliest，Error三个配置的行为是一样的。
但是如果连接的消费者所在的群组是一个新的群组时（也就是不存在offset记录），Latest，Earliest，Error三个配置表现出不一样的行为：
Latest：从连接到Kafka那一刻开始消费之后产生的消息，之前发布的消息不在消费，这也是默认的行为
Earliest：从offset最小值（如果消息全部有效的话，那就是最开头）处开始消费，也就是说会消费连接到Kafka之前发布的消息
Error：简单暴力的抛出异常
2、生产消息需要序列化，消费消息就需要反序列化了，我们需要提供一个实现了IDeserializer<TValue>接口的类型，比如上面的例子采用Json序列化：
public class KafkaConverter : IDeserializer<object> {/// <summary> /// 反序列化字节数据成实体数据 /// </summary> /// <param name="data"></param> /// <param name="context"></param> /// <returns></returns> public object Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context) { if (isNull) return null; var json = Encoding.UTF8.GetString(data.ToArray()); try { return JsonConvert.DeserializeObject(json); } catch { return json; } } }
3、Kafka是发布/订阅方式的消息队列，Confluent.Kafka提供了两个订阅的方法：Subscribe和Assign
Subscribe：从一个或者多个topic订阅消息
Assign：从一个或者多个topic的指定partition中订阅消息
另外，Confluent.Kafka还提供了两个取消订阅的方法：Unsubscribe和Unassign
4、获取消息使用Consume方法，方法返回一个ConsumeResult<TKey, TValue>对象，我们要的消息就在这个对象中，它还包含offset等等其他信息。
另外，Consume方法会导致当前线程阻塞，直至有获取到消息可以消费，或者超时。
5、如果我们创建消费者时，设置了EnableAutoCommit=false，那么我们就需要手动调用Commit方法提交消息，切记。
完整的Demo例子
上面有提到，生产消息需要一个实现序列化消息接口的对象，而消费消息需要一个实现了反序列化接口的对象，这两者建议用同一个类实现，于是一个完整的实现类：
public class KafkaConverter : ISerializer<object>, IDeserializer<object> { /// <summary> /// 序列化数据成字节 /// </summary> /// <param name="data"></param> /// <param name="context"></param> /// <returns></returns> public byte[] Serialize(object data, SerializationContext context) { var json = JsonConvert.SerializeObject(data); return Encoding.UTF8.GetBytes(json); } /// <summary> /// 反序列化字节数据成实体数据 /// </summary> /// <param name="data"></param> /// <param name="context"></param> /// <returns></returns> public object Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context) { if (isNull) return null; var json = Encoding.UTF8.GetString(data.ToArray()); try { return JsonConvert.DeserializeObject(json); } catch { return json; } } }
一个完整的Demo例子如下：
static void Main(string[] args) { var bootstrapServers = "192.168.209.133:9092,192.168.209.134:9092,192.168.209.135:9092"; var group1 = "group.1"; var group2 = "group.2"; var topic = "test"; new Thread(() => { ConsumerConfig config = new ConsumerConfig(); config.BootstrapServers = bootstrapServers; config.GroupId = group1; config.AutoOffsetReset = AutoOffsetReset.Earliest; config.EnableAutoCommit = false; var builder = new ConsumerBuilder<string, object>(config); builder.SetValueDeserializer(new KafkaConverter());//设置反序列化方式 var consumer = builder.Build(); //consumer.Subscribe(topic);//订阅消息使用Subscribe方法 consumer.Assign(new TopicPartition(topic, new Partition(0)));//从指定的Partition订阅消息使用Assign方法 while (true) { var result = consumer.Consume(); Console.WriteLine($"{group1} recieve message:{result.Message.Value}"); consumer.Commit(result);//手动提交，如果上面的EnableAutoCommit=true表示自动提交，则无需调用Commit方法 } }).Start(); new Thread(() => { ConsumerConfig config = new ConsumerConfig(); config.BootstrapServers = bootstrapServers; config.GroupId = group2; config.AutoOffsetReset = AutoOffsetReset.Earliest; config.EnableAutoCommit = false; var builder = new ConsumerBuilder<string, object>(config); builder.SetValueDeserializer(new KafkaConverter());//设置反序列化方式 var consumer = builder.Build(); //consumer.Subscribe(topic);//订阅消息使用Subscribe方法 consumer.Assign(new TopicPartition(topic, new Partition(1)));//从指定的Partition订阅消息使用Assign方法 while (true) { var result = consumer.Consume(); Console.WriteLine($"{group2} recieve message:{result.Message.Value}"); consumer.Commit(result);//手动提交，如果上面的EnableAutoCommit=true表示自动提交，则无需调用Commit方法 } }).Start(); int index = 0; ProducerConfig config = new ProducerConfig(); config.BootstrapServers = bootstrapServers; var builder = new ProducerBuilder<string, object>(config); builder.SetValueSerializer(new KafkaConverter());//设置序列化方式 var producer = builder.Build(); while (true) { Console.Write("请输入消息:"); var line = Console.ReadLine(); int partition = index % 3; var topicPartition = new TopicPartition(topic, new Partition(partition)); producer.Produce(topicPartition, new Message<string, object>() { Key = "Test", Value = line }); index++; } }
封装使用
这里做一个简单的封装，使用几个常用的配置以方便使用，当然，还是要使用nuget安装 Confluent.Kafka 和 Newtonsoft.Json两个包，具体几个类如下：
public abstract class KafkaBaseOptions { /// <summary> /// 服务器地址 /// </summary> public string[] BootstrapServers { get; set; } }
public class KafkaConsumer : IDisposable { ConsumerBuilder<string, object> builder; List<IConsumer<string, object>> consumers; bool disposed = false; /// <summary> /// kafka服务节点 /// </summary> public string BootstrapServers { get; private set; } /// <summary> /// 群组 /// </summary> public string GroupId { get; private set; } /// <summary> /// 是否允许自动提交（enable.auto.commit） /// </summary> public bool EnableAutoCommit { get; set; } = false; /// <summary> /// 异常事件 /// </summary> public event Action<object, Exception> ErrorHandler; /// <summary> /// 统计事件 /// </summary> public event Action<object, string> StatisticsHandler; /// <summary> /// 日志事件 /// </summary> public event Action<object, KafkaLogMessage> LogHandler; public KafkaConsumer(string groupId, params string[] bootstrapServers) { if (bootstrapServers == null || bootstrapServers.Length == 0) { throw new Exception("at least one server must be assigned"); } this.GroupId = groupId; this.BootstrapServers = string.Join(",", bootstrapServers); this.consumers = new List<IConsumer<string, object>>(); } #region Private /// <summary> /// 创建消费者生成器 /// </summary> private void CreateConsumerBuilder() { if (disposed) { throw new ObjectDisposedException(nameof(KafkaConsumer)); } if (builder == null) { lock (this) { if (builder == null) { ConsumerConfig config = new ConsumerConfig(); config.BootstrapServers = BootstrapServers; config.GroupId = GroupId; config.AutoOffsetReset = AutoOffsetReset.Earliest; config.EnableAutoCommit = EnableAutoCommit; //config.EnableAutoOffsetStore = true; //config.IsolationLevel = IsolationLevel.ReadCommitted; //config.MaxPollIntervalMs = 10000; //List<KeyValuePair<string, string>> config = new List<KeyValuePair<string, string>>(); //config.Add(new KeyValuePair<string, string>("bootstrap.servers", BootstrapServers)); //config.Add(new KeyValuePair<string, string>("group.id", GroupId)); //config.Add(new KeyValuePair<string, string>("auto.offset.reset", "earliest")); //config.Add(new KeyValuePair<string, string>("enable.auto.commit", EnableAutoCommit.ToString().ToLower())); //config.Add(new KeyValuePair<string, string>("max.poll.interval.ms", "10000")); //config.Add(new KeyValuePair<string, string>("session.timeout.ms", "10000")); //config.Add(new KeyValuePair<string, string>("isolation.level", "read_uncommitted")); builder = new ConsumerBuilder<string, object>(config); Action<Delegate, object> tryCatchWrap = (@delegate, arg) => { try { @delegate?.DynamicInvoke(arg); } catch { } }; builder.SetErrorHandler((p, e) => tryCatchWrap(ErrorHandler, new Exception(e.Reason))); builder.SetStatisticsHandler((p, e) => tryCatchWrap(StatisticsHandler, e)); builder.SetLogHandler((p, e) => tryCatchWrap(LogHandler, new KafkaLogMessage(e))); builder.SetValueDeserializer(new KafkaConverter()); } } } } /// <summary> /// 内部处理消息 /// </summary> /// <param name="consumer"></param> /// <param name="cancellationToken"></param> /// <param name="action"></param> private void InternalListen(IConsumer<string, object> consumer, CancellationToken cancellationToken, Action<RecieveResult> action) { try { var result = consumer.Consume(cancellationToken); if (!cancellationToken.IsCancellationRequested) { CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(); if (!EnableAutoCommit && result != null) { cancellationTokenSource.Token.Register(() => { consumer.Commit(result); }); } action?.Invoke(result == null ? null : new RecieveResult(result, cancellationTokenSource)); } } catch { } } /// <summary> /// 验证消费主题和分区 /// </summary> /// <param name="subscribers"></param> private void CheckSubscribers(params KafkaSubscriber[] subscribers) { if (subscribers == null || subscribers.Length == 0) { throw new InvalidOperationException("subscriber cann\'t be empty"); } if (subscribers.Any(f => string.IsNullOrEmpty(f.Topic))) { throw new InvalidOperationException("topic cann\'t be empty"); } } /// <summary> /// 设置监听主题 /// </summary> /// <param name="consumer"></param> private void SetSubscribers(IConsumer<string, object> consumer, params KafkaSubscriber[] subscribers) { var topics = subscribers.Where(f => f.Partition == null).Select(f => f.Topic).ToArray(); var topicPartitions = subscribers.Where(f => f.Partition != null).Select(f => new TopicPartition(f.Topic, new Partition(f.Partition.Value))).ToArray(); if (topics.Length > 0) { consumer.Subscribe(topics); } if (topicPartitions.Length > 0) { consumer.Assign(topicPartitions); } } /// <summary> /// 创建一个消费者 /// </summary> /// <param name="listenResult"></param> /// <param name="subscribers"></param> /// <returns></returns> private IConsumer<string, object> CreateConsumer(ListenResult listenResult, params KafkaSubscriber[] subscribers) { if (disposed) { throw new ObjectDisposedException(nameof(KafkaConsumer)); } CheckSubscribers(subscribers); CreateConsumerBuilder(); var consumer = builder.Build(); listenResult.Token.Register(() => { consumer.Dispose(); }); SetSubscribers(consumer, subscribers); consumers.Add(consumer); return consumer; } #endregion #region Listen /// <summary> /// 监听一次并阻塞当前线程，直至有消息获取或者取消获取 /// </summary> /// <param name="topics"></param> public RecieveResult ListenOnce(params string[] topics) { return ListenOnce(topics.Select(f => new KafkaSubscriber() { Partition =
全部评论
请发表评论