ZY.KAFKA 1.0.4

KAFKA

nuget引用组件 :ZY.KAFKA

配置

	{
        "Kafka": [
            {
                "Name": "Default",
                "BootstrapServers": "IP:Port",
                "GroupId": "分组",
                "SecurityProtocol": "SaslPlaintext", //SaslPlaintext， Plaintext, Ssl, SaslSsl
                "SaslMechanism": "ScramSha256", // Gssapi, Plain, ScramSha256, ScramSha512,  OAuthBearer
                "SaslUsername": "账号",
                "SaslPassword": "密码",
                "AutoOffsetReset": "Earliest",  // Latest, Earliest, Error
                "Topic": [
                    "主题1",
                    "主题2",
                    "主题3"
                ]
            },{
                "Name": "Name2",
                "BootstrapServers": "IP:Port",
                "GroupId": "分组",
                "SecurityProtocol": "SaslPlaintext", //SaslPlaintext， Plaintext, Ssl, SaslSsl
                "SaslMechanism": "ScramSha256", // Gssapi, Plain, ScramSha256, ScramSha512,  OAuthBearer
                "SaslUsername": "账号",
                "SaslPassword": "密码",
                "AutoOffsetReset": "Earliest",  // Latest, Earliest, Error
                "Topic": [
                    "主题1",
                    "主题2",
                    "主题3"
                ]
            }
        ]

    }

添加Startup

1、在ConfigureServices 添加代码

	 
    services.AddKafka(configuration);

2、在Configure添加代码

	
    app.UseKafka();

调用

    //接收消息处理业务
    internal class AppService
    {
        private readonly IServiceProvider _serviceProvider;
        private readonly ILogger _logger;

        public AppService(IServiceProvider serviceProvider, ILoggerFactory logger)
        {
            _serviceProvider = serviceProvider;
            _logger = logger.CreateLogger(this.GetType().Name);
        }

        /// <summary>
        /// 接收消息
        /// </summary>
        /// <param name="input"></param>
        public async Task Pull()
        {
            var service =  _serviceProvider.GetService<IKafkaClient>();
            service.AddEvent("Default", (result,msg) => //Default为kafka配置集合下Name值
            {
                _logger.LogInformation($"接收送信息：{msg}");
                //处理你的业务
                    
            });
            
            service.AddEvent("Name2", (result,msg) => //Name2为kafka配置集合下Name值
            {
                _logger.LogInformation($"接收送信息：{msg}");
                //处理你的业务
                    
            });
        }
    }

    //系统启用时调用 注：必须在 app.UseKafka() 调用前
    using (var scope = app.ApplicationServices.CreateScope())
    {
        scope.ServiceProvider.GetService<AppService>().Pull();
    }