一般在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！以上就完成了多线程多主题的消息发送。
相关文章：
消息队列 Kafka 的基本知识及 .NET Core 客户端
.net Kafka.Client多个Consumer Group对Topic消费不能完全覆盖研究总结（一）
.net Kafka.Client多个Consumer Group对Topic消费不能完全覆盖研究总结（二）
原文地址：http://www.cnblogs.com/zhangs1986/p/7285525.html
.NET社区新闻，深度好文，微信中搜索dotNET跨平台或扫描二维码关注
