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

一般在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;        }    }

然后在初始化的代码中替换Producer producer = new Producer(kfkip);为 Producer producer = SingleProduct.GetInstance;

OK!以上就完成了多线程多主题的消息发送。

(下载iPhone或Android应用“经理人分享”,一个只为职业精英人群提供优质知识服务的分享平台。不做单纯的资讯推送,致力于成为你的私人智库。)