JiShe.CollectBus/modules/JiShe.CollectBus.Kafka.Test/Program.cs

// See https://aka.ms/new-console-template for more information
using BenchmarkDotNet.Configs;
using BenchmarkDotNet.Running;
using Confluent.Kafka;
using DeviceDetectorNET.Parser.Device;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Kafka.AdminClient;
using JiShe.CollectBus.Kafka.Consumer;
using JiShe.CollectBus.Kafka.Producer;
using JiShe.CollectBus.Kafka.Test;
using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Serilog;
using System.Diagnostics;
using System.Reflection;
using System.Reflection.PortableExecutable;
using System.Text.Json;

#region 基准测试
//var summary = BenchmarkRunner.Run<KafkaProduceBenchmark>();
//Console.WriteLine("压测完成");
//return;
#endregion 基准测试


var host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        // 构建配置
        var config = new ConfigurationBuilder()
            .SetBasePath(Directory.GetCurrentDirectory())
            .AddJsonFile("appsettings.json")
            .Build();
        // 直接读取配置项
        var greeting = config["Kafka:ServerTagName"];
        Console.WriteLine(greeting);  // 输出: Hello, World!


        // 创建服务容器
        //var services = new ServiceCollection();
        // 注册 IConfiguration 实例
        services.AddSingleton<IConfiguration>(config);

        // 初始化日志
        Log.Logger = new LoggerConfiguration()
                  .ReadFrom.Configuration(config)  // 从 appsettings.json 读取配置
                  .CreateLogger();

        // 配置日志系统
        services.AddLogging(logging =>
        {
            logging.ClearProviders();
            logging.AddSerilog();
        });
        services.Configure<KafkaOptionConfig>(config.GetSection("Kafka"));

        services.AddSingleton<IAdminClientService, AdminClientService>();
        services.AddSingleton<IProducerService, ProducerService>();
        services.AddSingleton<IConsumerService, ConsumerService>();
        services.AddTransient<KafkaSubscribeTest>();

    })
    .ConfigureConsoleAppBuilder(appBuilder =>
    {
        
    })
    .Build();


  await host.StartAsync();
  var appBuilder = host.Services.GetRequiredService<IApplicationBuilder>();
  appBuilder.ApplicationServices.UseKafkaSubscribe();


// 构建ServiceProvider
//var serviceProvider = services.BuildServiceProvider();

// 获取日志记录器工厂
var loggerFactory = host.Services.GetRequiredService<ILoggerFactory>();
var logger = loggerFactory.CreateLogger<Program>();
logger.LogInformation("程序启动");
var adminClientService = host.Services.GetRequiredService<IAdminClientService>();
var configuration = host.Services.GetRequiredService<IConfiguration>();
string topic = "test-topic";
//await adminClientService.DeleteTopicAsync(topic);
// 创建 topic
//await adminClientService.CreateTopicAsync(topic, configuration.GetValue<int>(CommonConst.NumPartitions), 3);

var consumerService = host.Services.GetRequiredService<IConsumerService>();
//var kafkaOptions = host.Services.GetRequiredService<IOptions<KafkaOptionConfig>>();
//await consumerService.SubscribeAsync<object>(topic, (message) =>
//{
//    try
//    {
//        logger.LogInformation($"消费消息：{message}");
//        return Task.FromResult(true);

//    }
//    catch (ConsumeException ex)
//    {
//        // 处理消费错误
//        logger.LogError($"kafka消费异常:{ex.Message}");
//    }
//    return Task.FromResult(false);
//}, "default");

//Stopwatch stopwatch = Stopwatch.StartNew(); 

//for (int i = 0; i < 3; i++)
//{
//    await consumerService.SubscribeBatchAsync<dynamic>(topic, (message) =>
//    {
//        try
//        {
//            int index = 0;
//            logger.LogInformation($"消费消息_{index}消费总数：{message.Count()}：{JsonSerializer.Serialize(message)}");
//            return Task.FromResult(true);

//        }
//        catch (ConsumeException ex)
//        {
//            // 处理消费错误
//            logger.LogError($"kafka消费异常:{ex.Message}");
//        }
//        return Task.FromResult(false);
//    });
//}
//stopwatch.Stop();
//Console.WriteLine($"耗时: {stopwatch.ElapsedMilliseconds} 毫秒,{stopwatch.ElapsedMilliseconds/1000} 秒");

var producerService = host.Services.GetRequiredService<IProducerService>();
//int num = 840;
//while (num <= 900)
//{
//    //await producerService.ProduceAsync(topic, new TestTopic { Topic = topic, Val = i });
//    await producerService.ProduceAsync<string>(topic, num.ToString());
//    num++;
//}
await Task.Factory.StartNew(async() => {
    int num = 0;
    while (true)
    {
        //await producerService.ProduceAsync(topic, new TestTopic { Topic = topic, Val = i });
        await producerService.ProduceAsync<string>(topic, num.ToString());
        num++;
    }
});
Console.WriteLine("\n按Esc键退出");
while (true)
{
    var key = Console.ReadKey(intercept: true); // intercept:true 隐藏按键显示

    if (key.Key == ConsoleKey.Escape)
    {
        await host.StopAsync();
        Console.WriteLine("\n程序已退出");
        break;
    }
}
(host.Services as IDisposable)?.Dispose();


public class TestTopic
{
    public string Topic { get; set; }
    public int Val { get; set; }
}