【Kafka】基于Windows环境的Kafka有关环境(scala+zookeeper+kafka+可视化工具)搭建、以及使用.NET环境开发的案例代码与演示

admin

撰写于 2022年 01月 22 日

 /// 

/// Description: Kafka生产者发布服务
/// CreateTime: 2022/1/21 19:35:27
/// Author: Wesky
/// 

public class PublishService: IPublishService
{
public async Task PublishAsync(string broker, string topicName, TMessage message) where TMessage : class
{
var config = new ProducerConfig
{
BootstrapServers = broker, // kafka服务集群，例如  "192.168.0.1:9092,192.168.0.2:9092"   或者单机 "192.168.0.1:9092"
Acks = Acks.All,
MessageSendMaxRetries = 3, // 发送失败重试的次数
};
using (var producer = new ProducerBuilder(config).Build())
{
try
{
string data = Newtonsoft.Json.JsonConvert.SerializeObject(message);
var sendData = new Message { Key = Guid.NewGuid().ToString("N"), Value = data};
var report =  await producer.ProduceAsync(topicName, sendData);
Console.WriteLine($"消息 >>>>>： {data} \r\n发送到：{report.TopicPartitionOffset}");
}
catch (ProduceException ex)
{
Console.WriteLine($"消息发送失败>>>>>：\r\n Code= {ex.Error.Code} >>> \r\nError= {ex.Message}");
}
}
}
}

/// 

/// Description: kafka 消费者订阅服务
/// CreateTime: 2022/1/21 19:36:25
/// Author: Wesky
/// 

public class SubscribeService: ISubscribeService
{

        /// 

/// 消费者服务核心代码
/// 

/// 
/// 消费者配置信息
/// 主题集合
/// 
/// 
/// 
public async Task SubscribeAsync(ConsumerConfig config, IEnumerable topics,  Action func, CancellationToken cancellationToken) where TMessage : class
{
const int commitPeriod = 1;
using (var consumer = new ConsumerBuilder(config)
.SetErrorHandler((_, e) =>
{
Console.WriteLine($"消费错误 >>>>>: {e.Reason}");
})
.SetStatisticsHandler((_, json) =>
{
Console.WriteLine($"************************************************");
})
.SetPartitionsAssignedHandler((c, partitionList) =>
{
string partitions = string.Join(", ", partitionList);
Console.WriteLine($"分配的分区 >>>>> : {partitions}");
})
.SetPartitionsRevokedHandler((c, partitionList) =>
{
string partitions = string.Join(", ", partitionList);
Console.WriteLine($"回收的分区 >>>>> : {partitions}");
})
.Build())
{
consumer.Subscribe(topics);
try
{
while (true)
{
try
{
var consumeResult = consumer.Consume(cancellationToken);
if (consumeResult.IsPartitionEOF)
{
continue;
}
if(consumeResult?.Offset % commitPeriod == 0){
try
{
var result = JsonConvert.DeserializeObject(consumeResult.Message?.Value);
func(result); // 消费消息
}
catch (Exception ex)
{
Console.WriteLine($"消费业务处理失败: {ex.Message}");
}
try
{
consumer.Commit(consumeResult);  // 手动提交

                                    Console.WriteLine($"消费者消费完成，已提交 ");
}
catch (KafkaException e)
{
Console.WriteLine($"提交错误 >>>>> : {e.Error.Reason}");
}
}
}
catch (ConsumeException e)
{
Console.WriteLine($"消费错误>>>>> : {e.Error.Reason}");
}
}
}
catch (Exception e)
{
Console.WriteLine($"其他错误 >>>>> :{e.Message}");
consumer.Close();
}
}
await Task.CompletedTask;
}
}

  [Route("api/[controller]/[action]")]
[ApiController]
public class ProducerController : ControllerBase
{
IPublishService _service = null;
public ProducerController(IPublishService publishService)
{
_service = publishService;
}
[HttpPost]
public IActionResult SendMessage(string broker,string topicName,string message)
{
_service.PublishAsync(broker, topicName, message);
return Ok();
}
}