可用的Kafka .NET客户端

confluent-kafka-dotnet

基于Confluent.Kafka的Sample

要完成本文示例，首先得有一个启动好的Kafka Broker服务。

部署Kafka环境

参考Kafka学习二-基于Docker搭Kafka

创建项目

创建一个My.Kafka.Core.Demo控制台项目，安装Confluent.Kafka组件

  1. Install-Package Confluent.Kafka
  2. Install-Package System.Text.Json

编写KafkaService

  1. namespace My.Kafka.Core.Demo;
  3. public interface IKafkaService
  4. {
  5.     /// <summary>
  6.     /// 发布
  7.     /// </summary>
  8.     /// <typeparam name="T"></typeparam>
  9.     /// <param name="topicName"></param>
  10.     /// <param name="message"></param>
  11.     /// <returns></returns>
  12.     Task PublishAsync<T>(string topicName, T message) where T : class;
  14.     /// <summary>
  15.     /// 订阅
  16.     /// </summary>
  17.     /// <typeparam name="T"></typeparam>
  18.     /// <param name="topics"></param>
  19.     /// <param name="messageFunc"></param>
  20.     /// <param name="cancellationToken"></param>
  21.     /// <returns></returns>
  22.     Task SubscribeAsync<T>(IEnumerable<string> topics, Action<T> messageFunc, CancellationToken cancellationToken = default) where T : class;
  23. }
  1. using Confluent.Kafka;
  2. using System.Text.Json;
  4. namespace My.Kafka.Core.Demo;
  6. public class KafkaService : IKafkaService
  7. {
  8.     public static string KAFKA_SERVERS = "127.0.0.1:9091";
  10.     public async Task PublishAsync<T>(string topicName, T message) where T : class
  11.     {
  12.         var config = new ProducerConfig
  13.         {
  14.             BootstrapServers = KAFKA_SERVERS,
  15.             BatchSize = 16384, // 修改批次大小为16K
  16.             LingerMs = 20 // 修改等待时间为20ms
  17.         };
  18.         using var producer = new ProducerBuilder<string, string>(config).Build();
  19.         await producer.ProduceAsync(topicName, new Message<string, string>
  20.         {
  21.             Key = Guid.NewGuid().ToString(),
  22.             Value = JsonSerializer.Serialize(message)
  23.         });
  24.     }
  26.     public async Task SubscribeAsync<T>(IEnumerable<string> topics, Action<T> messageFunc, CancellationToken cancellationToken = default) where T : class
  27.     {
  28.         var config = new ConsumerConfig
  29.         {
  30.             BootstrapServers = KAFKA_SERVERS,
  31.             GroupId = "Consumer",
  32.             EnableAutoCommit = false, // 禁止AutoCommit
  33.             Acks = Acks.Leader, // 假设只需要Leader响应即可
  34.             AutoOffsetReset = AutoOffsetReset.Earliest // 从最早的开始消费起
  35.         };
  37.         using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
  38.         {
  39.             consumer.Subscribe(topics);
  40.             try
  41.             {
  42.                 while (true)
  43.                 {
  44.                     try
  45.                     {
  46.                         var consumeResult = consumer.Consume(cancellationToken);
  47.                         Console.WriteLine($"Consumed message '{consumeResult.Message?.Value}' at: '{consumeResult?.TopicPartitionOffset}'.");
  48.                         if (consumeResult.IsPartitionEOF)
  49.                         {
  50.                             Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} 已经到底了：{consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
  51.                             continue;
  52.                         }
  54.                         T messageResult = null;
  55.                         try
  56.                         {
  57.                             //messageResult = JsonConvert.DeserializeObject<T>(consumeResult.Message.Value);
  58.                             if(consumeResult!=null&& consumeResult.Message!=null)
  59.                                 messageResult = JsonSerializer.Deserialize<T>(consumeResult.Message.Value);
  60.                         }
  61.                         catch (Exception ex)
  62.                         {
  63.                             var errorMessage = $" - {DateTime.Now:yyyy-MM-dd HH:mm:ss}【Exception 消息反序列化失败，Value：{consumeResult.Message.Value}】 ：{ex.StackTrace?.ToString()}";
  64.                             Console.WriteLine(errorMessage);
  65.                             messageResult = null;
  66.                         }
  68.                         if (messageResult != null/* && consumeResult.Offset % commitPeriod == 0*/)
  69.                         {
  70.                             messageFunc(messageResult);
  71.                             try
  72.                             {
  73.                                 consumer.Commit(consumeResult);
  74.                             }
  75.                             catch (KafkaException e)
  76.                             {
  77.                                 Console.WriteLine(e.Message);
  78.                             }
  79.                         }
  80.                     }
  81.                     catch (ConsumeException e)
  82.                     {
  83.                         Console.WriteLine($"Consume error: {e.Error.Reason}");
  84.                     }
  85.                 }
  86.             }
  87.             catch (OperationCanceledException)
  88.             {
  89.                 Console.WriteLine("Closing consumer.");
  90.                 consumer.Close();
  91.             }
  92.         }
  94.         await Task.CompletedTask;
  95.     }
  96. }
  1. namespace My.Kafka.Core.Demo;
  3. public class EventData
  4. {
  5.     public string TopicName { get; set; }
  7.     public string Message { get; set; }
  9.     public DateTime EventTime { get; set; }
  10. }

编写Producer

新建一个Console项目，暂且命名为：My.Kafka.Demo.Producer，其主体内容如下：

  2. using My.Kafka.Core.Demo;
  4. KafkaService.KAFKA_SERVERS = "kafka1:9091,kafka2:9092,kafka3:9093";
  5. var kafkaService = new KafkaService();
  6. for (int i = 0; i < 50; i++)
  7. {
  8.     var eventData = new EventData
  9.     {
  10.         TopicName = "testtopic",
  11.         Message = $"This is a message from Producer, Index : {i + 1}",
  12.         EventTime = DateTime.Now
  13.     };
  14.     await kafkaService.PublishAsync<EventData>(eventData.TopicName, eventData);
  15. }
  16. Console.WriteLine("发布结束!");
  17. Console.ReadKey();

编写Consumer

新建一个Console项目，暂且命名为：My.Kafka.Demo.Consumer，其主体内容如下：

  1. using My.Kafka.Core.Demo;
  3. KafkaService.KAFKA_SERVERS = "kafka1:9091,kafka2:9092,kafka3:9093";
  4. var kafkaService = new KafkaService();
  5. var topics = new List<string> { "testtopic" };
  6. await kafkaService.SubscribeAsync<EventData>(topics, (eventData) =>
  7. {
  8.     Console.WriteLine($" - {eventData.EventTime: yyyy-MM-dd HH:mm:ss} 【{eventData.TopicName}】- > 已处理");
  9. });

测试Pub/Sub效果

将Producer和Consumer两个项目都启动起来(ctrl+F5)，可以看到当Consumer消费完50条消息并一一确认之后，Producer这边就算发布结束。
1663214475378.png

基于CAP项目的Sample

模拟场景说明

假设我们有两个微服务，一个是Catalog微服务，一个是Basket微服务，当Catalog微服务产生了Product价格更新的事件，就会将其发布到Kafka，Basket微服务作为消费者就会订阅这个消息然后更新购物车中对应商品的最新价格。
Kafkaafka学习三-.NET Core示例 - 图2

Catalog.WebAPI

新建一个名称为Catalog.WebAPI的ASP.NET Core WebAPI项目，然后分别安装以下组件：

  1. Install-Package DotNetCore.CAP
  2. Install-Package DotNetCore.CAP.MongoDB
  3. Install-Package DotNetCore.CAP.Kafka
  4. Install-Package AutoMapper

注入CAP

  1. //注入CAP
  2. builder.Services.AddCap(x =>
  3. {
  4.     x.UseMongoDB("mongodb://account:password@mongodb-server:27017/products?authSource=admin");
  5.     x.UseKafka("kafka1:9091,kafka2:9092,kafka3:9093");
  6. });
  1. namespace Catalog.WebAPI;
  3. public class Product
  4. {
  5.     public string Id { get; set; }
  7.     public string Name { get; set; }
  8.     public decimal Price { get; set; }
  10.     public int Stock { get; set; }
  12.     public string Introduction { get; set; }
  13. }
  1. namespace Catalog.WebAPI;
  3. public class ProductDTO
  4. {
  5.     public string Id { get; set; }
  6.     public string Name { get; set; }
  7.     public decimal Price { get; set; }
  8.     public string Introduction { get; set; }
  9. }

创建ProductController,实现一个Update产品价格的接口，在其中通过CapPublisher完成发布消息到Kafka。

  1. using AutoMapper;
  2. using Catalog.WebAPI;
  3. using DotNetCore.CAP;
  4. using Microsoft.AspNetCore.Mvc;
  5. using static Confluent.Kafka.ConfigPropertyNames;
  7. var builder = WebApplication.CreateBuilder(args);
  9. // Add services to the container.
  10. // Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
  11. builder.Services.AddEndpointsApiExplorer();
  13. //注入CAP
  14. builder.Services.AddCap(x =>
  15. {
  16.     //x.UseMongoDB("mongodb://account:password@mongodb-server:2717/products?authSource=admin");
  17.     x.UseMongoDB("mongodb://localhost:2717/products?authSource=admin");
  18.     x.UseKafka("kafka1:9091,kafka2:9092,kafka3:9093");
  19. });
  21. builder.Services.AddSwaggerGen();
  23. var app = builder.Build();
  25. // Configure the HTTP request pipeline.
  26. if (app.Environment.IsDevelopment())
  27. {
  28.     app.UseSwagger();
  29.     app.UseSwaggerUI();
  30. }
  32. app.UseHttpsRedirection();
  34. IList<Product> GetProductList()
  35. {
  36.     IList<Product> products = new List<Product>
  37.         {
  38.             new Product { Id = "0001", Name = "电动牙刷A", Price = 99.90M,  Introduction = "暂无介绍" },
  39.             new Product { Id = "0002", Name = "电动牙刷B", Price = 199.90M,  Introduction = "暂无介绍" },
  40.             new Product { Id = "0003", Name = "洗衣机A", Price = 2999.90M,  Introduction = "暂无介绍" },
  41.             new Product { Id = "0004", Name = "洗衣机B", Price = 3999.90M,  Introduction = "暂无介绍" },
  42.             new Product { Id = "0005", Name = "电视机A", Price = 1899.90M,  Introduction = "暂无介绍" },
  43.         };
  44.     return products;
  45. }
  48. app.MapGet("/Get", () =>
  49. {
  50.     //return _mapper.Map<IList<ProductDTO>>(GetProductList());
  51.     return GetProductList();
  52. });
  54. app.MapPut("/UpdatePrice", async (ICapPublisher _publisher,string id, decimal newPrice) =>
  55. {
  56.     // 业务代码
  57.     var product = GetProductList().FirstOrDefault(p => p.Id == id);
  58.     product.Price = newPrice;
  60.     // 发布消息
  61.     await _publisher.PublishAsync("ProductPriceChanged",
  62.         new ProductDTO { Id = product.Id, Name = product.Name, Price = product.Price });
  64. });
  66. app.Run();
  68. internal record WeatherForecast(DateTime Date, int TemperatureC, string? Summary)
  69. {
  70.     public int TemperatureF => 32 + (int)(TemperatureC / 0.5556);
  71. }

Basket.WebAPI

参照Catalog API项目创建ASP.NET Core WebAPI项目，并安装对应组件。

  1. Install-Package DotNetCore.CAP
  2. Install-Package DotNetCore.CAP.MongoDB
  3. Install-Package DotNetCore.CAP.Kafka
  4. Install-Package AutoMapper

注入CAP

  1. //注入CAP
  2. builder.Services.AddCap(x =>
  3.                         {
  4.     x.UseMongoDB("mongodb://account:password@mongodb-server:27017/products?authSource=admin");
  5.     x.UseKafka("kafka1:9091,kafka2:9092,kafka3:9093");
  6. });

新建一个BasketController，用于订阅Kafka对应Topic：ProductPriceChanged 的消息。

  2. using DotNetCore.CAP;
  3. using Microsoft.AspNetCore.Mvc;
  5. namespace Basket.WebAPIC;
  7. [Route("[controller]")]
  8. [ApiController]
  9. public class BasketController : ControllerBase
  10. {
  11.     private static readonly IList<MyBasketDTO> Baskets = new List<MyBasketDTO>
  12.     {
  13.         new MyBasketDTO { UserId = "U001", Catalogs = new List<Catalog>
  14.             {
  15.                 new Catalog { Product = new ProductDTO { Id = "0001", Name = "电动牙刷A", Price = 99.90M }, Count = 2 },
  16.                 new Catalog { Product = new ProductDTO { Id = "0005", Name = "电视机A", Price = 1899.90M }, Count = 1 },
  17.             }
  18.         },
  19.         new MyBasketDTO { UserId = "U002", Catalogs = new List<Catalog>
  20.             {
  21.                 new Catalog { Product = new ProductDTO { Id = "0002", Name = "电动牙刷B", Price = 199.90M }, Count = 2 },
  22.                 new Catalog { Product = new ProductDTO { Id = "0004", Name = "洗衣机B", Price = 3999.90M }, Count = 1 },
  23.             }
  24.         }
  25.     };
  27.     [HttpGet]
  28.     public IList<MyBasketDTO> Get()
  29.     {
  30.         return Baskets;
  31.     }
  33.     [NonAction]
  34.     [CapSubscribe("ProductPriceChanged")]
  35.     public async Task RefreshBasketProductPrice(ProductDTO productDTO)
  36.     {
  37.         if (productDTO == null)
  38.             return;
  40.         foreach (var basket in Baskets)
  41.         {
  42.             foreach (var catalog in basket.Catalogs)
  43.             {
  44.                 if (catalog.Product.Id == productDTO.Id)
  45.                 {
  46.                     catalog.Product.Price = productDTO.Price;
  47.                     break;
  48.                 }
  49.             }
  50.         }
  52.         await Task.CompletedTask;
  53.     }
  54. }
  1. namespace Basket.WebAPI.Models;
  3. public class Catalog
  4. {
  5.     public ProductDTO Product { get; set; }
  6.     public int Count { get; set; }
  7. }
  1. namespace Basket.WebAPI.Models;
  3. public class MyBasketDTO
  4. {
  5.     public string UserId { get; set; }
  6.     public IList<Catalog> Catalogs { get; set; }
  7. }
  1. namespace Basket.WebAPI.Models;
  3. public class Product
  4. {
  5.     public string Id { get; set; }
  7.     public string Name { get; set; }
  8.     public decimal Price { get; set; }
  10.     public int Stock { get; set; }
  12.     public string Introduction { get; set; }
  13. }
  1. namespace Basket.WebAPI.Models;
  3. public class ProductDTO
  4. {
  5.     public string Id { get; set; }
  6.     public string Name { get; set; }
  7.     public decimal Price { get; set; }
  8.     public string Introduction { get; set; }
  9. }

测试效果

同时启动Catalog API 和 Basket API两个项目。
首先，通过Swagger在Basket API中查看所有用户购物车中的商品的价格，可以看到，0002的商品是199.9元。
Kafkaafka学习三-.NET Core示例 - 图3
然后，通过Swagger在Catalog API中更新Id为0002的商品的价格至499.9元。
Kafkaafka学习三-.NET Core示例 - 图4
1663232629763.png
最后，通过Swagger在Basket API中查看所有用户购物车中的商品的价格，可以看到，0002的商品已更新至499.9元。