NET中解决KafKa多线程发送多主题的问题-阿里云开发者社区

NET中解决KafKa多线程发送多主题的问题

2017-11-22 881

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介：

一般在KafKa消费程序中消费可以设置多个主题，那在同一程序中需要向KafKa发送不同主题的消息，如异常需要发到异常主题，正常的发送到正常的主题，这时候就需要实例化多个主题，然后逐个发送。

　　在NET中用RdKafka组件来做消息处理，在Nuget中引用。

　　在程序中初始化Producer，并创建多个Topic

        private string comtopic = "topic1";
        private string errtopic = "topic2";
        private string kfkip = "192.168.80.32:9092";
        Topic topic = null;
        Topic errTopic = null;

        public ExcuteFlow()
        {
            try
            {
                Producer producer = new Producer(kfkip);
                topic = producer.Topic(comtopic);
                errTopic = producer.Topic(errtopic);
            }
            catch (RdKafkaException ex)
            {
                LogHelper.Error("KafKa初始化KafKa异常 ", ex);
            }
            catch (Exception ex)
            {
                LogHelper.Error("KafKa初始化异常", ex);
            }

        }

　　在程序中发送其中一个主题：

          try
            {

                if (topic != null)
                {
                    byte[] datas = Encoding.UTF8.GetBytes(JsonHelper.ToJson(flowCommond));
                    Task<DeliveryReport> deliveryReport = topic.Produce(datas);
                    var unused = deliveryReport.ContinueWith(task =>
                    {
                        LogHelper.Info("内容：{flowCommond.ID} 发送到分区：{task.Result.Partition}, Offset 为: {task.Result.Offset}");
                    });
                }
                else
                {
                    throw new Exception("发送消息到KafKa topic 为空");
                }
            }
            catch (RdKafkaException ex)
            {
                LogHelper.Error("发送消息到KafKa  KafKa异常", ex);
            }
            catch (Exception ex)
            {
                LogHelper.Error("发送消息到KafKa异常", ex);
            }

　　flowCommond为要发送的对象内容，格式化为Json字符串再发送。

　　另一个主题一样处理。

　　这里实现一个线程里面发送多个主题，那下面实现多个线程中如何发送多个主题。

　　多线程中如果每个线程都new Producer(kfkip) 一次，那KafKa的连接很快会被占满。

　　那这里就用单例模式来解决这个问题，每次要用到Producer时检查一下是否已经存在Producer实例，若存在则直接用不用再生成。

    /// <summary>
    /// 单例模式的实现
    /// </summary>
    public class SingleProduct : Producer
    {
        // 定义一个静态变量来保存类的实例
        private static SingleProduct uniqueInstance;
        // 定义一个标识确保线程同步
        private static readonly object locker = new object();
        // 定义私有构造函数，使外界不能创建该类实例
        private SingleProduct(string brokerList) : base(brokerList)
        {
        }

        /// <summary>
        /// 定义公有方法提供一个全局访问点,同时你也可以定义公有属性来提供全局访问点
        /// </summary>
        /// <returns></returns>
        public static SingleProduct GetInstance()
        {
            // 当第一个线程运行到这里时，此时会对locker对象 "加锁"，
            // 当第二个线程运行该方法时，首先检测到locker对象为"加锁"状态，该线程就会挂起等待第一个线程解锁
            // lock语句运行完之后（即线程运行完之后）会对该对象"解锁"
            if (uniqueInstance == null)
            {
                lock (locker)
                {
                    // 如果类的实例不存在则创建，否则直接返回
                    if (uniqueInstance == null)
                    {
                        string kfkip = System.Configuration.ConfigurationManager.AppSettings["KfkIP"];

                        try
                        {
                            uniqueInstance = new SingleProduct(kfkip);
                            LogHelper.Error("单例模式 实例化 SingleProduct");
                        }
                        catch (RdKafkaException ex)
                        {
                            LogHelper.Error("单例模式 KafKa初始化KafKa异常 ", ex);
                        }
                        catch (Exception ex)
                        {
                            LogHelper.Error("单例模式 KafKa初始化异常", ex);
                        }
                    }
                }
            }

            return uniqueInstance;
        }
    }