【C#】Kafka for .NET 入坑记录

【C#】Kafka for .NET 入坑记录

一、有关Kafka

Kafka 是一个分布式、多副本、基于Zookeeper的消息流处理平台。Kafka 最初由Linkedin开发，为了办理 data pipeline 问题。2011年募捐给 Apache基金会，后来成为Apache的顶级开源项目。
Kafka 的未来：及时数据流平台。（出自饶军）
1-1、基本概念

• 消息：Kafka 中的数据单元，也被称为记录。
  
• 批次：消息会分批次写入Kafka，批次就是指一组消息。
  
• 主题：消息的种类，也可以说一个主题代表一类消息。
  
• 分区：物理上最小的存储单元。一个主题可以分为多个分区，每个分区内消息是有序的。
  
• 生产者：生产消息，发送到主题。
  
• 消费者：从主题获取消息，处理消息。
  
• 消费者群组：由一个或多个消费者组成的群体。在同一个消费组中，一个消息只能被一个消息者消息。
  
• 偏移量：一个不断自增的整数值，分区中一条消息的唯一标示符。可体现消费者在分区的位置。
  
• broker：一个独立的Kafka服务器。
  
• broker集群：由一个或多个broker组成的集群。
  
• 副本：消息的备份。Kafka定义了两类副本（领导者副本、追随者副本）。
  
• 重平衡：其个消费者实例挂掉后，其他消息者实例自动重新分配的过程。
  

相干词汇：

1. 消息 - message
2. 分区 - partition
3. 主题 - topic
4. 生产者 - producer
5. 消费者 - consumer
6. 消费者群组 - consumer group
7. 偏移量 - consumer offset
8. 缓存代理 - broker
9. 副本 - replica
10. 领导者副本 - leader replica
11. 追随者副本 - follower replica
12. 重平衡 - rebalance

复制代码

1-2、Kafka发行版本：

• Apache Kafak：也称社区版 Kafka。
  
• Confluent Kafka：Confluent 公司提供的 Kafka。
  2014年，Kafka的3个首创人 Jay Kreps、Naha Narkhede 和 Jun Rao 离开 LinkedIn，创立了Confluent公司，专注于提供基于 Kafka 的企业流处理办理方案。Confluent Kafka 提供了一些 Apache Kafka 没有的高级特性，如跨数据中央备份、Schema 注册中央以及集群监控工具等。
  
• Cloudera/Hortonworks Kafka：Cloudera 提供的 CDH Kafka 和 Hortonworks 提供的 HDP Kafka。集成了 Apache Kafka，通过便捷化的界面操作将 Kafka 的安装、运维、管理、监控全部统一在控制台中。
  

二、Technology Libraries

2-1、Confluent.Kafka

引用库

1. <PackageReference Include="Confluent.Kafka" Version="2.8.0" />

复制代码

1. const string topic = "test";
2. const string host = "localhost:9092";
4. #region 消费者
6. var t1 = Task.Run(async () =>
7. {
8.          var cfg = new ConsumerConfig()
9.          {
10.                  GroupId = $"test-consumer-group-console",
11.                  BootstrapServers = host,
12.                  AutoOffsetReset = AutoOffsetReset.Earliest,
13.                  EnableAutoCommit = true  // 自动提交
14.          };
16.          using var builder = new ConsumerBuilder<Ignore, string>(cfg).Build();
17.          builder.Subscribe(topic);
19.          var cts = new CancellationTokenSource();
20.          Console.CancelKeyPress += (s, e) =>
21.          {
22.                  e.Cancel = true;
23.                  cts.Cancel();
24.          };
25.          try
26.          {
27.                  while (true)
28.                  {
29.                          try
30.                          {
31.                                  var consumer = builder.Consume(cts.Token);
32.                                  Console.WriteLine($"Message: {consumer.Message.Value} received from {consumer.TopicPartitionOffset}");
34.                                  // builder.Commit(consumer); // 手动提交
35.                          }
36.                          catch (ConsumeException e)
37.                          {
38.                                  Console.WriteLine($"Error occured: {e.Error.Reason}");
39.                          }
41.                          await Task.Yield();
42.                  }
43.          }
44.          catch (OperationCanceledException ex)
45.          {
46.                  builder.Close();
47.                  Console.WriteLine($"ex: {ex}");
48.          }
49. });
51. #endregion 消费者
53. #region 生产者
55. var t2 = Task.Run(async () =>
56. {
57.         var cfg = new ProducerConfig
58.         {
59.                 BootstrapServers = host,
60.         };
61.         var cts = new CancellationTokenSource();
62.         Console.CancelKeyPress += (s, e) =>
63.         {
64.                 e.Cancel = true;
65.                 cts.Cancel();
66.         };
68.         // 生产者
69.         using var producer = new ProducerBuilder<Null, string>(cfg).Build();
70.         while (true)
71.         {
72.                 if (cts.IsCancellationRequested) break;
74.                 try
75.                 {
76.                         await producer.ProduceAsync(topic, new Message<Null, string>
77.                         {
78.                                 Value = JsonConvert.SerializeObject(new KafkaMessage
79.                                 {
80.                                         Ts = DateTimeOffset.Now,
81.                                         Text = new Faker("en").Address.FullAddress()
82.                                 })
83.                         });
84.                 }
85.                 catch (ProduceException<Null, KafkaMessage> ex)
86.                 {
87.                         Console.WriteLine($"ex: {ex.Error.Reason}");
88.                 }
90.                 await Task.Delay(TimeSpan.FromMilliseconds(200));
91.         }
92. });
94. #endregion 生产者
96. Console.ReadLine();
97. await Task.WhenAll(t1, t2);

复制代码

2-2、DotNetCore.CAP.Kafka

CAP 是一种处理分布式事件的办理方案，能够保证任务环境下变乱消息都不会丢失。可作 EventBus 来使用。
引用库

1. <PackageReference Include="DotNetCore.CAP.InMemoryStorage" Version="8.3.2" />
2. <PackageReference Include="DotNetCore.CAP.Kafka" Version="8.3.2" />
3. <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="9.0.0" />
4. <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="9.0.0" />

复制代码

1. var services = new ServiceCollection();
2. services.AddLogging(config =>
3. {
4.    config.SetMinimumLevel(LogLevel.Debug);
5.    config.AddConsole();
6. });
8. services.AddCap(config =>
9. {
10.    config.UseInMemoryStorage();
11.    config.UseKafka(opt =>
12.    {
13.            opt.Servers = "localhost:9092";
14.           
15.            // 自定义Headers, cap-msg-id 和 cap-msg-name 必选项
16.            opt.CustomHeadersBuilder = (kr, sp) => new List<KeyValuePair<string, string>>
17.            {
18.                    new KeyValuePair<string, string>("cap-msg-id",Guid.NewGuid().ToString()),
19.                    new KeyValuePair<string, string>("cap-msg-name","test"),
20.                    new KeyValuePair<string, string>("kafka.offset",kr.Offset.ToString()),
21.            };
22.    });
23.    config.FailedRetryInterval = 10; // 重试的间隔时间，默认是60s
24.    config.DefaultGroupName = "test-group";
25. });
27. services.AddTransient<ICapSubscribe, SubscriberService>();
29. var sp = services.BuildServiceProvider();
31. var cts = new CancellationTokenSource();
32. Console.CancelKeyPress += (s, e) =>
33. {
34.    e.Cancel = true;
35.    cts.Cancel();
36. };
38. // 启动CAP
39. var bootstrap = sp.GetRequiredService<IBootstrapper>();
40. await bootstrap.BootstrapAsync(cts.Token);
41. await Task.Delay(500);
43. // 生产者
44. ICapPublisher capBus = sp.GetRequiredService<ICapPublisher>();
45. while (true)
46. {
47.    if (cts.IsCancellationRequested) break;
49.    // 推送消息
50.    capBus.Publish("test", new KafkaMessage
51.    {
52.            Ts = DateTimeOffset.Now,
53.            Text = new Faker().Address.FullAddress()
54.    });
55.           
56.    await Task.Delay(TimeSpan.FromMilliseconds(200));
57. }
59. await bootstrap.DisposeAsync();
60. Console.ReadLine();

复制代码

消费者

1. internal class SubscriberService : ICapSubscribe
2. {
3.    [CapSubscribe("test")]
4.    public void ReceiveMessage(KafkaMessage msg, [FromCap] CapHeader header)
5.    {
6.            if (msg != null)
7.            {
8.                    Console.WriteLine($"Message: {msg.Ts} | {msg.Text}");
9.            }
10.    }
11. }

复制代码

2-3、MassTransit.Kafka

MassTransit 基于消息驱动的分布式应用框架，可作为分布式应用的消息总线，也可以用作单体应用的变乱总线。
引用库

1. <PackageReference Include="MassTransit.Kafka" Version="8.3.4" />
2. <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="9.0.0" />

复制代码

1. var services = new ServiceCollection();
3. services.AddMassTransit(config =>
4. {
5.   const string topic = "test";
6.   const string groupId = "test-consumer-group";
7.   const string host = "localhost:9092";
9.   config.UsingInMemory((context, cfg) => cfg.ConfigureEndpoints(context));
11.   config.AddRider(rider =>
12.   {
13.           // 注册生产者
14.           rider.AddProducer<KafkaMessage>(topic);
15.           // 注册消费者
16.           rider.AddConsumer<KafkaMessageConsumer>();
18.           // Kafka连接配置
19.           rider.UsingKafka((context, k) =>
20.           {
21.                   k.Host(host);
22.                   // 配置 topic
23.                   k.TopicEndpoint<KafkaMessage>(topic, groupId, e =>
24.                   {
25.                           e.AutoOffsetReset = AutoOffsetReset.Earliest;
26.                           e.ConfigureConsumer<KafkaMessageConsumer>(context);
27.                           e.CreateIfMissing();
28.                   });
29.           });
30.   });
31. });
33. var sp = services.BuildServiceProvider();
35. var cts = new CancellationTokenSource();
36. Console.CancelKeyPress += (s, e) =>
37. {
38.   e.Cancel = true;
39.   cts.Cancel();
40. };
42. var bus = sp.GetRequiredService<IBusControl>();
43. bus.Start(TimeSpan.FromSeconds(10.0));
44. try
45. {
46.   // 生产者
47.   var producer = sp.GetRequiredService<ITopicProducer<KafkaMessage>>();
48.   while (true)
49.   {
50.           if (cts.IsCancellationRequested) break;
51.           try
52.           {
53.                   // 推送消息
54.                   string value = new Faker().Address.FullAddress();
55.                   await producer.Produce(new KafkaMessage
56.                   {
57.                           Ts = DateTimeOffset.Now,
58.                           Text = value
59.                   });
61.                   await Task.Delay(TimeSpan.FromMilliseconds(200));
62.           }
63.           catch (ProduceException<Ignore, KafkaMessage> pex)
64.           {
65.                   Console.WriteLine(pex.Error.Reason);
66.           }
67.   }
68. }
69. catch (OperationCanceledException cce)
70. {
71. }
72. finally
73. {
74.   await bus.StopAsync();
75. }
77. Console.ReadLine();

复制代码

消费者

1. internal class KafkaMessageConsumer : IConsumer<KafkaMessage>
2. {
3. public Task Consume(ConsumeContext<KafkaMessage> context)
4. {
5.         var ctx = (context.ReceiveContext as KafkaReceiveContext<Ignore, KafkaMessage>);
6.         Console.WriteLine($"Message: {context.Message.Ts} | {context.Message.Text}, Offset: {ctx?.Offset}");
8.         return Task.CompletedTask;
9. }
10. }

复制代码

2-4、Streamiz.Kafka.Net

Kafka Streams 流式计算

1. // kafka连接配置
2. var config = new StreamConfig<StringSerDes, StringSerDes>();
3. config.ApplicationId = "test-app";
4. config.BootstrapServers = "localhost:9092";
6. StreamBuilder builder = new StreamBuilder();
8. builder.Stream<string, string>("test") // 输入 topic (订阅 topic = test)
9.         // 定义处理过程
10.         .MapValues((value, context) =>
11.         {
12.                 return $"{value} ---- 1111 ---- {context.Offset}";
13.         })
14.         // 定义处理过程
15.         .MapValues((value, context) =>
16.         {
17.                 return $"{value} ---- 2222 ---- {context.Offset}";
18.         })
19.         // 定义处理过程
20.         // ...
21.         // 定义处理过程
22.         .To("test-output");  // 输出 topic (把处理结束发到 topic = test-output)
24. Topology t = builder.Build();
25. KafkaStream stream = new KafkaStream(t, config);
27. await stream.StartAsync();
29. Console.ReadLine();
31. stream.Dispose();

复制代码

输入：a
输出：a ---- 1111 ---- 4910 ---- 2222 ---- 4910

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。