【.NET】Kafka消息队列介绍，使用Confluent.Kafka集成Kafka消息队列
一、Kafka介绍
kafka是一种高吞吐量、分布式、可扩展的消息中间件系统，最初由LinkedIn公司开发。随着不断的发展，在最新的版本中它定义为分布式的流处理平台，现在在大数据应用中也是十分广泛。
它可以处理大量的实时数据流，被广泛应用于日志收集、事件处理、流处理、消息队列等场景。
Kafka的架构包含producer（生产者）、consumer（消费者）、broker（代理服务器）等组件。生产者可以将消息发送到Kafka集群，消费者可以从Kafka集群订阅消息并进行处理，而broker则是消息的中转服务器，负责存储和转发消息。
Kafka的特点包括：
- 高吞吐量：Kafka可以处理海量的数据流，支持每秒百万级别的消息处理。
- 可扩展性：Kafka的集群可以根据需要进行水平扩展，从而提高系统的性能和容量。
- 可靠性：Kafka支持多副本机制，可以保证数据的可靠性和高可用性。
- 灵活性：Kafka支持多种消息格式和协议，可以与各种系统和工具进行集成。
- Kafka是一个开源的项目，已经成为了Apache软件基金会的顶级项目.
Kafka & 核心概念
接着，我们看下它的核心概念,这些概念都很重要，在后边的学习中都会遇到，概念一定要搞明白，对于理解Kafka的工作原理和使用方法非常重要。不然学习起来比较懵, 下面一起看一下核心概念:
Topic
Topic是消息的逻辑容器，用于对消息进行分类和存储。在Kafka中，消息会被发布到指定的topic中，并且可以被一个或多个消费者订阅。Topic是Kafka的核心概念之一，是实现消息传递的基础。
Producer
Producer是消息的生产者，用于向指定的topic中发送消息。Producer负责将消息发送到Kafka集群中的broker节点，并且可以在发送消息时指定消息的key，以便Kafka将消息分配到指定的partition中。
Consumer
Consumer是消息的消费者，用于从指定的topic中接收消息。Consumer负责从Kafka集群中的broker节点获取消息，并且可以指定从哪个partition中获取消息。消费者可以以不同的方式进行消息消费，例如批量消费、轮询消费等。
Broker
Broker是Kafka集群中的一个节点，用于存储和管理消息。Broker是Kafka的核心组件之一，负责接收和处理生产者发送的消息，并将其存储到磁盘中，同时还负责将消息转发给消费者。
Partition
Partition是Kafka中实现数据分片的机制，一个topic可以被分成多个partition，每个partition都是一个有序的消息队列。消息在被发送到一个topic时，会被根据指定的key进行hash计算，然后被分配到对应的partition中。
Offset
Offset是Kafka中的一个重要概念，用于标识每个消息在一个partition中的位置。每个partition都有一个唯一的offset值，消费者可以根据offset来获取指定位置的消息。Kafka还提供了一种特殊的topic，称为__consumer_offsets，用于存储消费者消费的位置信息。
参考：https://zhuanlan.zhihu.com/p/612327585
二、C#引用Confluent.Kafka.dll实现kafka消息队列的实际开发例子
1、配置文件
"KafkaConfig": {
"BootstrapServers": "",
"SaslUsername": "",
"SaslPassword": ""
},
2、配置类： BusinessOptionsSetting.cs
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace IntegratedPlatform.Domain.Dtos.OptionSetting
{
/// <summary>
/// 业务配置
/// </summary>
public class BusinessOptionsSetting
{
/// <summary>
/// kafka 配置
/// </summary>
public KafkaConfig KafkaConfig { get; set; }
}
/// <summary>
/// kafka 配置
/// </summary>
public class KafkaConfig
{
/// <summary>
/// 服务端配置
/// </summary>
public string BootstrapServers { get; set; }
/// <summary>
/// Sasl用户名
/// </summary>
public string SaslUsername { get; set; }
/// <summary>
/// Sasl密码
/// </summary>
public string SaslPassword { get; set; }
}
}
3、kafka服务类： KafkaServices.cs
using Confluent.Kafka;
using IntegratedPlatform.Domain.Dtos.OptionSetting;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using NPOI.XWPF.UserModel;
using Serilog;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Security.Cryptography.Xml;
using System.Text;
using System.Threading.Tasks;
using static Confluent.Kafka.ConfigPropertyNames;
namespace IntegratedPlatform.Infrastructure.Kafka
{
/// <summary>
/// kafka服务类
/// </summary>
public class KafkaServices
{
private BusinessOptionsSetting options;
public KafkaServices(IOptions<BusinessOptionsSetting> _options)
{
options = _options.Value;
}
/// <summary>
/// 发送消息至指定主题
/// </summary>
/// <param name="topicName">主题名称</param>
/// <param name="message">消息内容</param>
/// <returns>异步任务</returns>
public async Task PublishMessageAsync(string topicName, string message)
{
var config = new ProducerConfig
{
BootstrapServers = options.BootstrapServers,
EnableIdempotence = true, // 启用幂等性以防止重复发送
Acks = Acks.All, // 确保所有副本都收到消息
SecurityProtocol = SecurityProtocol.SaslPlaintext,
SaslMechanism = SaslMechanism.ScramSha256,
SaslUsername = options.SaslUsername,
SaslPassword = options.SaslPassword,
BatchNumMessages = 1, // 每个批次发送一条消息
AllowAutoCreateTopics = true, // 允许自动创建主题
MessageSendMaxRetries = 3, // 最大重试次数
};
using (var producer = new ProducerBuilder<string, string>(config)
//.SetValueSerializer(new CustomStringSerializer<string>())
.Build())
{
try
{
var deliveryReport = await producer.ProduceAsync(topicName, new Message<string, string> { Key = "1", Value = message });
// Console.WriteLine($"delivered to: {deliveryReport.TopicPartitionOffset}");
}
catch (ProduceException<string, string> e)
{
Log.Error($"failed to deliver message: {e.Message} [{e.Error.Code}]");
}
}
}
/// <summary>
/// 从指定主题订阅消息
/// </summary>
/// <param name="topics"></param>
/// <param name="messageFunc"></param>
/// <param name="cancellationToken"></param>
/// <param name="groupId"></param>
/// <returns></returns>
public void SubscribeAsync(IEnumerable<string> topics, Action<string> messageFunc, CancellationToken cancellationToken, string groupId)
{
var config = new ConsumerConfig
{
BootstrapServers = options.KafkaConfig.BootstrapServers,
GroupId = groupId,
EnableAutoCommit = false,
StatisticsIntervalMs = 5000,
SecurityProtocol = SecurityProtocol.SaslPlaintext,
SaslMechanism = SaslMechanism.ScramSha256,
SaslUsername = options.KafkaConfig.SaslUsername,
SaslPassword = options.KafkaConfig.SaslPassword,
AutoOffsetReset = AutoOffsetReset.Earliest,
EnablePartitionEof = true,
};
//提交偏移量的时候,也可以批量去提交
const int commitPeriod = 1;
using (var consumer = new ConsumerBuilder<Ignore, string>(config)
.SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}") )
.SetStatisticsHandler((_, json) =>
{
Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} > Kafka消息监听中..");
})
.SetPartitionsAssignedHandler((c, partitions) =>
{
string partitionsStr = string.Join(", ", partitions);
Console.WriteLine($" - 分配的 kafka 分区: [{string.Join(", ", partitions)}]");
})
.SetPartitionsRevokedHandler((c, partitions) =>
{
//自定义存储偏移量
//1.每次消费完成，把相应的分区id和offset写入到mysql数据库存储
//2.从指定分区和偏移量开始拉取数据
//分配的时候调用
Console.WriteLine($" - 回收了 kafka 的分区:[{string.Join(", ", partitions)}]");
})
//.SetValueDeserializer(new CustomStringIDeserializer<T>())
.Build())
{
consumer.Subscribe(topics);
try
{
while (true)
{
try
{
var consumeResult = consumer.Consume(cancellationToken);
if (consumeResult.IsPartitionEOF)
{
Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} 已经到底了：{consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
continue;
}
Console.WriteLine($"Kafka接收到消息 at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");
string messageResult = consumeResult.Message.Value;
if (!string.IsNullOrEmpty(messageResult) /*&& consumeResult.Offset % commitPeriod == 0*/)
{
messageFunc(messageResult);
try
{
consumer.Commit(consumeResult);
}
catch (KafkaException e)
{
Log.Error($"Commit error: {e.Error.Reason}");
}
}
}
catch (ConsumeException e)
{
Log.Error($"Consume error: {e.Error.Reason}");
}
}
}
catch (OperationCanceledException ex)
{
Log.Error($"Closing consumer.{ex.Message}");
consumer.Close();
}
}
}
}
}
4、使用 kafka
/// <summary>
/// 用户消息订阅（主题：MSG_UCENTER_PUBLISH_ZYYD）
/// </summary>
/// <param name="o"></param>
public void MsgUserSubscribe(object? o)
{
var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
e.Cancel = true;
cts.Cancel();
};
_kafkaServices.SubscribeAsync(new List<string>() { EConstant.KafkaTopicName.USER }, async (eventData) =>
{
//Console.WriteLine($" - {eventData}】- > 已处理");
try
{
var messageEntity = new KafkaMessage()
{
MType = 1,
AcType = 1,
TopicName = EConstant.KafkaTopicName.USER,
Content = eventData,
Status = 0
};
var id = kafkaMessageRepository.AddReturnIdentity(messageEntity);
messageEntity.Id = id;
JObject jobject = JObject.Parse(eventData);
}
catch (Exception e)
{
Log.Error($"failed to deliver message: {e.Message}");
}
}, cts.Token, EConstant.KafkaGroupId.USER);
}