[C#] Kafka 生产者和消费者实现_随笔

[C#] Kafka 生产者和消费者实现

一、背景

二、简介

2.1 Kafka

三、实现

3.1 创建项目

3.2 安装Package

3.3 代码实现

3.3.1 实现消息序列化和反序列化的接口

3.3.2 实现辅助类

3.3.3 实现公共的配置类

3.3.4 实现生产者

3.3.5 消费者

3.3.6 添加一个日志类

3.3.7 打包发布Nuget

四、应用

4.1搭建验证程式

4.1.1 创建一个控制台应用程序

4.1.2 引用Kafka组件

4.1.3 使用生产者和消费者

五、说明

一、背景

由于公司加强对员工 *** 作记录的审查和追踪，程式需要对员工的进行存档记录。由于并发巨大，使用传统的直连DB进行存储的方式对DB造成巨大压力，也导致程序响应缓慢，降低了用户体验。结合DBA推动Kafka，开发Kafka调用组件，使各应用程式快速集成Kafka，实现高并发的消息队列处理方案。

二、简介 2.1 Kafka

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

特性：

Kafka [1]

是一种高吞吐量 [2]

的分布式发布订阅消息系统，有如下特性：

通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。

高吞吐量 [2] ：即使是非常普通的硬件Kafka也可以支持每秒数百万 [2] 的消息。

支持通过Kafka服务器和消费机集群来分区消息。

支持Hadoop并行数据加载。 [3]

Kafka通过官网发布了最新版本3.0.0 [4]

[6]

来源：百度百科

三、实现 3.1 创建项目

先创建一个类库的项目，我们把它叫做KafkaHelper，

3.2 安装Package

使用Nuget安装以下Package

-----------------------------------------------------------------------------------------------------

Confluent.Kafka

Newtonsoft.Json ——用于消息序列化和反序列化

-----------------------------------------------------------------------------------------------------

以上包有依赖项，需要全部引入，引用完成后的packages.config文件如下：

3.3 代码实现

代码中注释比较全面，我就不多说了，有问题可以评论或私信找我

3.3.1 实现消息序列化和反序列化的接口

注意：如果报错

CS0619 “ReadOnlySpan”已过时:“Types with embedded references are not supported in this version of your compiler.”

请使用VS2019+版本开发组件

using Confluent.Kafka;
using Newtonsoft.Json;
using System;
using System.Text;

namespace KafkaHelper
{
    class KafkaConverter : ISerializer
    {
        /// 
        /// 序列化数据成字节
        /// 
        /// 
        /// 
        /// 
        public byte[] Serialize(T data, SerializationContext context)
        {
            var json = JsonConvert.SerializeObject(data);
            return Encoding.UTF8.GetBytes(json);
        }
    }
    class KafkaDConverter : IDeserializer
    {
        /// 
        /// 反序列化字节数据成实体数据
        /// 
        /// 
        /// 
        /// 
        /// 
        public T Deserialize(ReadOnlySpan data, bool isNull, SerializationContext context)
        {
            if (isNull) return default(T);

            var json = Encoding.UTF8.GetString(data.ToArray());
            try
            {
                return JsonConvert.DeserializeObject(json);
            }
            catch
            {
                return default(T);
            }
        }
    }
}

3.3.2 实现辅助类

辅助类主要用于实现一些常见公共的方法

using System.Reflection;
using System.Text.Regularexpressions;

namespace KafkaHelper
{
    /// 
    /// 辅助类
    /// 
    public class Helper
    {
        /// 
        /// 获取当前应用程式名称(仅控制台应用程序和Windows应用程序可用)
        /// 
        /// 
        public static string GetApplicationName()
        {
            try
            {
                return Assembly.GetEntryAssembly().GetName().Name;
            }
            catch
            {
                return "Kafka_Test";
            }
        }

        /// 
        /// 获取服务器名称
        /// 
        /// 
        public static string GetServerName()
        {
            return System.Net.Dns.GetHostName();
        }

        /// 
        /// 获取服务器IP
        /// 
        /// 
        public static string GetServerIp()
        {
            System.Net.IPHostEntry ips = System.Net.Dns.GetHostEntry(System.Net.Dns.GetHostName());
            foreach (var ip in ips.AddressList)
            {
                if (Regex.IsMatch(ip.ToString(), @"^10.((25[0-5]|2[0-4]d|1d{2}|d?d).){2}(25[0-5]|2[0-4]d|1d{2}|d?d)$"))
                {
                    return ip.ToString();
                };
            }
            return "127.0.0.1";
        }

        ///   
        /// 将c# DateTime时间格式转换为Unix时间戳格式(毫秒级)  
        ///   
        /// long  
        public static long GetTimeStamp()
        {
            System.DateTime time = System.DateTime.Now;
            long t = (time.Ticks - 621356256000000000) / 10000;
            return t;
        }
    }
}

3.3.3 实现公共的配置类

该类的设置主要是便于不同身份的继承和使用

using Confluent.Kafka;

namespace KafkaHelper
{
    /// 
    /// 配置类
    /// 
    public class KafkaConfig
    {
        /// 
        /// 构造配置类
        /// 
        protected KafkaConfig()
        {
            ProducerConfig = new ProducerConfig()
            {
                BootstrapServers = "xxx.xxx.xxx.xxx:9092"
            };

            ConsumerConfig = new ConsumerConfig()
            {
                BootstrapServers = "xxx.xxx.xxx.xxx:9092",
                GroupId = "group.1",
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableAutoCommit = false
            };
        }

        /// 
        /// 消费者配置文件
        /// 
        public ConsumerConfig ConsumerConfig;

        /// 
        /// 生产者配置文件
        /// 
        public ProducerConfig ProducerConfig;
    }
}

3.3.4 实现生产者

我这边生产者使用了单例模式(静态变量实现单例)

using Confluent.Kafka;
using System;

namespace KafkaHelper
{
    /// 
    /// 生产者 Message.Key的数据类型为string、Message.Value的数据类型为string
    /// 
    public class Producer : Producer
    {
        private Producer()
        {
        }
    }

    /// 
    /// 生产者
    /// 
    /// Message.Key 的数据类型
    /// Message.Value 的数据类型
    public class Producer : KafkaConfig
    {

        private static readonly Producer producer = new Producer();

        /// 
        /// 构造生产者 (私有)
        /// 
        protected Producer()
        {
        }

        /// 
        /// 获取生产者实例
        /// 
        /// 
        public static Producer GetProducer
        {
            get
            {
                return producer;
            }
        }

        /// 
        ///  Kafka地址(包含端口号) 默认为 127.0.0.1:9092
        /// 
        public string Servers
        {
            get
            {
                return ProducerConfig.BootstrapServers;
            }
            set
            {
                ProducerConfig.BootstrapServers = value;
            }
        }

        /// 
        /// 主题 默认为当前程式名称(Web应用 默认为 Kafka_Test)
        /// 
        public string Topic { get; set; } = Helper.GetApplicationName();

        private IProducer _producer;

        /// 
        /// 启动
        /// 
        /// 是否成功
        public bool Start()
        {
            string msg;
            return Start(out msg);
        }

        /// 
        /// 启动
        /// 
        /// 启动结果说明
        /// 是否成功
        public bool Start(out string Message)
        {
            try
            {
                var producerBuilder = new ProducerBuilder(ProducerConfig);
                producerBuilder.SetValueSerializer(new KafkaConverter());//设置序列化方式
                _producer = producerBuilder.Build();
                Message = "启动成功";
                return true;
            }
            catch (Exception ex)
            {
                Message = ex.ToString();
                return false;
            }
        }

        /// 
        /// 生产(使用配置的Topic)
        /// 
        /// Message.Key
        /// Message.Value
        public void Produce(TKey Key, TValue Value)
        {
            Produce(Key, Value, Topic);
        }

        /// 
        /// 生产(使用传入的Topic)
        /// 
        /// Message.Key
        /// Message.Value
        /// 主题
        public void Produce(TKey Key, TValue Value, string Topic)
        {
            _producer.Produce(Topic, new Message() { Key = Key, Value = Value });
        }

    }
}

3.3.5 消费者

消费者实现三种消费方式： 1.订阅回调模式 2.批量消费模式 3.单笔消费模式

using Confluent.Kafka;
using System;
using System.Collections.Generic;

namespace KafkaHelper
{
    /// 
    /// 消费者 (Message.Key的数据类型为string、Message.Value的数据类型为string）
    /// 
    public class Consumer : Consumer { }

    /// 
    /// 消费者
    /// 
    /// Message.Key 的数据类型
    /// Message.Value 的数据类型
    public class Consumer : KafkaConfig
    {
        /// 
        ///  Kafka地址(包含端口号) 默认为 127.0.0.1:9092
        /// 
        public string Servers
        {
            get
            {
                return ConsumerConfig.BootstrapServers;
            }
            set
            {
                ConsumerConfig.BootstrapServers = value;
            }
        }

        /// 
        /// 消费者群组 默认为当前程式名称(Web应用 默认为 ISD_Kafka)
        /// 
        public string GroupId
        {
            get
            {
                return ConsumerConfig.GroupId;
            }
            set
            {
                ConsumerConfig.GroupId = value;
            }
        }

        /// 
        /// 自动提交 默认为 false
        /// 
        public bool EnableAutoCommit
        {
            get
            {
                return ConsumerConfig.EnableAutoCommit ?? false;
            }
            set
            {
                ConsumerConfig.EnableAutoCommit = value;
            }
        }

        private IConsumer _consumer;

        /// 
        /// 启动
        /// 
        /// 是否成功
        public bool Start()
        {
            string msg;
            return Start(out msg);
        }

        /// 
        /// 启动
        /// 
        /// 启动结果说明
        /// 是否成功
        public bool Start(out string Message)
        {
            try
            {
                var Builder = new ConsumerBuilder(ConsumerConfig);
                Builder.SetValueDeserializer(new KafkaDConverter());//设置反序列化方式
                _consumer = Builder.Build();
                Message = "启动成功";
                return true;
            }
            catch (Exception ex)
            {
                Message = ex.ToString();
                return false;
            }
        }

        /// 
        /// 消费(持续订阅)
        /// 
        /// 回调函数,若配置为非自动提交(默认为否),则通过回调函数的返回值判断是否提交
        /// 主题
        public void Consume(Func, bool> Func, string Topic)
        {
            _consumer.Subscribe(Topic);

            while (true)
            {
                try
                {
                    var result = _consumer.Consume();
                    if (Func(result))
                    {
                        if (!(bool)ConsumerConfig.EnableAutoCommit)
                        {
                            _consumer.Commit(result);//手动提交，如果上面的EnableAutoCommit=true表示自动提交，则无需调用Commit方法
                        }
                    }
                }
                catch
                {
                }
            }
        }

        /// 
        /// 单次消费(消费出当前Kafka缓存的所有数据,并持续监听 300ms,如无新数据生产,则返回(最多一次消费 100条)
        /// 
        /// 主题
        /// 持续监听时间,单位ms 默认值:300ms
        /// 最多单次消费行数 默认值:100行
        /// 待消费数据
        public List> ConsumeOnce(string Topic, int TimeOut = 300, int MaxRow = 100)
        {
            _consumer.Subscribe(Topic);

            List> Res = new List>();
            while (true)
            {
                try
                {
                    var result = _consumer.Consume(TimeSpan.FromMilliseconds(TimeOut));
                    if (result == null)
                    {
                        break;
                    }
                    else
                    {
                        Res.Add(result);
                        _consumer.Commit();//手动提交，如果上面的EnableAutoCommit=true表示自动提交，则无需调用Commit方法
                    }
                    if (Res.Count > MaxRow)
                    {
                        break;
                    }

                }
                catch
                {
                }
            }
            return Res;
        }

        /// 
        /// 单行消费
        /// 
        /// 主题
        /// 持续监听时间,单位ms 默认值:300ms
        /// 待消费数据
        public ConsumeResult ConsumeOneRow(string Topic, int TimeOut = 300)
        {
            _consumer.Subscribe(Topic);
            try
            {
                var result = _consumer.Consume(TimeSpan.FromMilliseconds(TimeOut));
                if (result != null)
                {
                    _consumer.Commit();//手动提交，如果上面的EnableAutoCommit=true表示自动提交，则无需调用Commit方法
                }
                return result;

            }
            catch
            {
                return null;
            }
        }

    }
}

3.3.6 添加一个日志类

使用日志类，便于日志记录和标准统一

using System;

namespace KafkaHelper
{
    /// 
    /// 默认日志类
    /// 
    public class LogModel
    {
        /// 
        /// 构造默认日志类(设置默认值 ServerIp,ServerName,TimeStamp,TimeStamp,ApplicationVersion)
        /// 
        public LogModel()
        {
            ServerIp = Helper.GetServerIp();
            ServerName = Helper.GetServerName();
            TimeStamp = System.DateTime.Now;
            ApplicationName = Helper.GetApplicationName();
            ApplicationVersion = "V1.0.0";
        }

        /// 
        /// 程式名称(默认获取当前程式名称,Web应用 默认为 ISD_Kafka)
        /// 
        public string ApplicationName { get; set; }

        /// 
        /// 程式版本(默认为V1.0.0)
        /// 
        public string ApplicationVersion { get; set; }

        /// 
        /// 发生时间(默认为当前时间)
        /// 
        public DateTime TimeStamp { get; set; }

        /// 
        /// 开始时间
        /// 
        public DateTime BeginDate { get; set; }

        /// 
        /// 结束时间
        /// 
        public DateTime EndDate { get; set; }

        /// 
        /// 服务器IP(默认抓取当前服务器IP)
        /// 
        public string ServerIp { get; set; }

        /// 
        /// 服务器名称(默认抓取当前服务器名称)
        /// 
        public string ServerName { get; set; }

        /// 
        /// 客户端IP
        /// 
        public string ClientIp { get; set; }

        /// 
        /// 模块(页面路径)
        /// 
        public string Module { get; set; }

        /// 
        ///  *** 作人
        /// 
        public string Operator { get; set; }

        /// 
        ///  *** 作类型 如:Query,Add,Update,Delete,Export等,可自定义
        /// 
        public string OperationType { get; set; }

        /// 
        ///  *** 作状态 如:http请求使用200,404,503等,其他 *** 作 1:成功,0失败等 可自定义
        /// 
        public string Status { get; set; }

        /// 
        /// 其他信息
        /// 
        public string Message { get; set; }
    }
}

到这里我们的Kafka辅助组件已经开发完毕了

3.3.7 打包发布Nuget

公司内部搭建了Nuget服务器，使用 NuGet Package Explorer 打包组件，并发布到服务器上

四、应用 4.1搭建验证程式 4.1.1 创建一个控制台应用程序

4.1.2 引用Kafka组件

直接是用Nuget安装刚刚发布的组件

我这里把测试项目直接写在了同一个专案中，所以就直接引用了

4.1.3 使用生产者和消费者

using KafkaHelper;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApp1
{
    class Program
    {
        static void Main(string[] args)
        {
            var msg = "hello world by Younger";

            //开启新线程订阅消息
            Thread rad = new Thread(new ThreadStart(Read));
            rad.Start();

            //获取生产者
            var producer = KafkaHelper.Producer.GetProducer;
            //设置主题
            producer.Topic = "test1";
            //开启生产者
            producer.Start();

        inputMsg:
            Console.WriteLine($"[send] [{System.DateTime.Now.ToLongTimeString()}] value:{msg}");

            //生产消息
            producer.Produce("message", msg);

            Thread.Sleep(1000);
            Console.WriteLine("");
            Console.Write("Please input message:");
            msg = Console.ReadLine();
            goto inputMsg;
        }

        /// 
        /// 订阅回调模式
        /// 
        private static void Read()
        {
            //实例化消费者
            KafkaHelper.Consumer consumer = new KafkaHelper.Consumer();
            //开启消费者
            consumer.Start();
            //订阅消息
            consumer.Consume(r =>
            {
                Console.WriteLine($"[recieve] [{System.DateTime.Now.ToLongTimeString()}] value:{r.Message.Value}");
                return true;
            }, "test1");
        }

        /// 
        /// 批量消费模式
        /// 
        private static void Readonce()
        {
            //实例化消费者
            KafkaHelper.Consumer consumer = new KafkaHelper.Consumer();
            //开启消费者
            consumer.Start();

            while (true)
            {
                //设置指定时长消费一次数据
                Thread.Sleep(5000);

                //消费并返回当前数据集合
                var Res = consumer.ConsumeOnce("test1");

                Console.WriteLine();
                foreach (var n in Res)
                {
                    Console.WriteLine($"[recieve] [{System.DateTime.Now.ToLongTimeString()}] value:{n.Message.Value}");
                }
                Console.WriteLine();

            }
        }

        /// 
        /// 单笔消费模式
        /// 
        private static void ReadoneRow()
        {
            KafkaHelper.Consumer consumer = new KafkaHelper.Consumer();
            consumer.Start();
            var a = consumer.ConsumeOneRow("test1");
        }
    }
}

执行结果如下图所示

五、说明

生产者使用单例模式可以降低程式消耗，在使用时

建议Web项目使用Global.asax全局文件配置

示例：

using System;

namespace WebApplication3
{
    public class Global : System.Web.HttpApplication
    {
        protected void Application_Start(object sender, EventArgs e)
        {
            var producer = KafkaHelper.Producer.GetProducer;
            producer.Topic="test1";
            producer.Start();
        }
    }
}

控制台和Windows应用程式直接配置在主函数中

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/zaji/5669032.html

[C#] Kafka 生产者和消费者实现

发表评论

评论列表（0条）