Kafka基础教程（三）：C#使用Kafka消息队列

public abstract class KafkaBaseOptions
{
/// <summary>
/// 服务器地址
/// </summary>
public string[] BootstrapServers { get; set; }
}

public class KafkaConsumer : IDisposable
{
ConsumerBuilder<string, object> builder;
List<IConsumer<string, object>> consumers;
bool disposed = false; /// <summary>
/// kafka服务节点
/// </summary>
public string BootstrapServers { get; private set; }
/// <summary>
/// 群组
/// </summary>
public string GroupId { get; private set; }
/// <summary>
/// 是否允许自动提交（enable.auto.commit）
/// </summary>
public bool EnableAutoCommit { get; set; } = false; /// <summary>
/// 异常事件
/// </summary>
public event Action<object, Exception> ErrorHandler;
/// <summary>
/// 统计事件
/// </summary>
public event Action<object, string> StatisticsHandler;
/// <summary>
/// 日志事件
/// </summary>
public event Action<object, KafkaLogMessage> LogHandler; public KafkaConsumer(string groupId, params string[] bootstrapServers)
{
if (bootstrapServers == null || bootstrapServers.Length == 0)
{
throw new Exception("at least one server must be assigned");
} this.GroupId = groupId;
this.BootstrapServers = string.Join(",", bootstrapServers);
this.consumers = new List<IConsumer<string, object>>();
} #region Private
/// <summary>
/// 创建消费者生成器
/// </summary>
private void CreateConsumerBuilder()
{
if (disposed)
{
throw new ObjectDisposedException(nameof(KafkaConsumer));
} if (builder == null)
{
lock (this)
{
if (builder == null)
{
ConsumerConfig config = new ConsumerConfig();
config.BootstrapServers = BootstrapServers;
config.GroupId = GroupId;
config.AutoOffsetReset = AutoOffsetReset.Earliest;
config.EnableAutoCommit = EnableAutoCommit;
//config.EnableAutoOffsetStore = true;
//config.IsolationLevel = IsolationLevel.ReadCommitted;
//config.MaxPollIntervalMs = 10000; //List<KeyValuePair<string, string>> config = new List<KeyValuePair<string, string>>();
//config.Add(new KeyValuePair<string, string>("bootstrap.servers", BootstrapServers));
//config.Add(new KeyValuePair<string, string>("group.id", GroupId));
//config.Add(new KeyValuePair<string, string>("auto.offset.reset", "earliest"));
//config.Add(new KeyValuePair<string, string>("enable.auto.commit", EnableAutoCommit.ToString().ToLower()));
//config.Add(new KeyValuePair<string, string>("max.poll.interval.ms", "10000"));
//config.Add(new KeyValuePair<string, string>("session.timeout.ms", "10000"));
//config.Add(new KeyValuePair<string, string>("isolation.level", "read_uncommitted")); builder = new ConsumerBuilder<string, object>(config); Action<Delegate, object> tryCatchWrap = (@delegate, arg) =>
{
try
{
@delegate?.DynamicInvoke(arg);
}
catch { }
};
builder.SetErrorHandler((p, e) => tryCatchWrap(ErrorHandler, new Exception(e.Reason)));
builder.SetStatisticsHandler((p, e) => tryCatchWrap(StatisticsHandler, e));
builder.SetLogHandler((p, e) => tryCatchWrap(LogHandler, new KafkaLogMessage(e)));
builder.SetValueDeserializer(new KafkaConverter());
}
}
}
}
/// <summary>
/// 内部处理消息
/// </summary>
/// <param name="consumer"></param>
/// <param name="cancellationToken"></param>
/// <param name="action"></param>
private void InternalListen(IConsumer<string, object> consumer, CancellationToken cancellationToken, Action<RecieveResult> action)
{
try
{
var result = consumer.Consume(cancellationToken);
if (!cancellationToken.IsCancellationRequested)
{
CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
if (!EnableAutoCommit && result != null)
{
cancellationTokenSource.Token.Register(() =>
{
consumer.Commit(result);
});
}
action?.Invoke(result == null ? null : new RecieveResult(result, cancellationTokenSource));
}
}
catch { }
}
/// <summary>
/// 验证消费主题和分区
/// </summary>
/// <param name="subscribers"></param>
private void CheckSubscribers(params KafkaSubscriber[] subscribers)
{
if (subscribers == null || subscribers.Length == 0)
{
throw new InvalidOperationException("subscriber cann't be empty");
} if (subscribers.Any(f => string.IsNullOrEmpty(f.Topic)))
{
throw new InvalidOperationException("topic cann't be empty");
}
}
/// <summary>
/// 设置监听主题
/// </summary>
/// <param name="consumer"></param>
private void SetSubscribers(IConsumer<string, object> consumer, params KafkaSubscriber[] subscribers)
{
var topics = subscribers.Where(f => f.Partition == null).Select(f => f.Topic).ToArray();
var topicPartitions = subscribers.Where(f => f.Partition != null).Select(f => new TopicPartition(f.Topic, new Partition(f.Partition.Value))).ToArray(); if (topics.Length > 0)
{
consumer.Subscribe(topics);
} if (topicPartitions.Length > 0)
{
consumer.Assign(topicPartitions);
}
}
/// <summary>
/// 创建一个消费者
/// </summary>
/// <param name="listenResult"></param>
/// <param name="subscribers"></param>
/// <returns></returns>
private IConsumer<string, object> CreateConsumer(ListenResult listenResult, params KafkaSubscriber[] subscribers)
{
if (disposed)
{
throw new ObjectDisposedException(nameof(KafkaConsumer));
} CheckSubscribers(subscribers); CreateConsumerBuilder(); var consumer = builder.Build();
listenResult.Token.Register(() =>
{
consumer.Dispose();
}); SetSubscribers(consumer, subscribers); consumers.Add(consumer); return consumer;
}
#endregion #region Listen
/// <summary>
/// 监听一次并阻塞当前线程，直至有消息获取或者取消获取
/// </summary>
/// <param name="topics"></param>
public RecieveResult ListenOnce(params string[] topics)
{
return ListenOnce(topics.Select(f => new KafkaSubscriber() { Partition = null, Topic = f }).ToArray());
}
/// <summary>
/// 监听一次并阻塞当前线程，直至有消息获取或者取消获取
/// </summary>
/// <param name="subscribers"></param>
public RecieveResult ListenOnce(params KafkaSubscriber[] subscribers)
{
ListenResult listenResult = new ListenResult();
var consumer = CreateConsumer(listenResult, subscribers); RecieveResult result = null;
InternalListen(consumer, listenResult.Token, rr =>
{
result = rr;
});
return result;
}
/// <summary>
/// 异步监听一次
/// </summary>
/// <param name="topics"></param>
/// <returns></returns>
public async Task<RecieveResult> ListenOnceAsync(params string[] topics)
{
return await ListenOnceAsync(topics.Select(f => new KafkaSubscriber() { Partition = null, Topic = f }).ToArray());
}
/// <summary>
/// 异步监听一次
/// </summary>
/// <param name="subscribers"></param>
/// <returns></returns>
public async Task<RecieveResult> ListenOnceAsync(params KafkaSubscriber[] subscribers)
{
return await Task.Run(() =>
{
return ListenOnce(subscribers);
});
}
/// <summary>
/// 监听
/// </summary>
/// <param name="topics"></param>
/// <param name="action"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public void Listen(string[] topics, Action<RecieveResult> action = null, CancellationToken cancellationToken = default)
{
Listen(topics.Select(f => new KafkaSubscriber() { Partition = null, Topic = f }).ToArray(), action, cancellationToken);
}
/// <summary>
/// 监听
/// </summary>
/// <param name="subscribers"></param>
/// <param name="action"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public void Listen(KafkaSubscriber[] subscribers, Action<RecieveResult> action = null, CancellationToken cancellationToken = default)
{
ListenResult result = new ListenResult();
var consumer = CreateConsumer(result, subscribers);
cancellationToken.Register(() =>
{
result.Stop();
});
while (!result.Stoped)
{
InternalListen(consumer, result.Token, action);
}
}
/// <summary>
/// 异步监听
/// </summary>
/// <param name="topics"></param>
/// <param name="action"></param>
/// <returns></returns>
public async Task<ListenResult> ListenAsync(string[] topics, Action<RecieveResult> action = null)
{
return await ListenAsync(topics.Select(f => new KafkaSubscriber() { Partition = null, Topic = f }).ToArray(), action);
}
/// <summary>
/// 异步监听
/// </summary>
/// <param name="subscribers"></param>
/// <param name="action"></param>
/// <returns></returns>
public async Task<ListenResult> ListenAsync(KafkaSubscriber[] subscribers, Action<RecieveResult> action = null)
{
ListenResult result = new ListenResult();
new Task(() =>
{
var consumer = CreateConsumer(result, subscribers);
while (!result.Stoped)
{
InternalListen(consumer, result.Token, action);
}
}).Start();
return await Task.FromResult(result);
}
#endregion
/// <summary>
/// 释放资源
/// </summary>
public void Dispose()
{
disposed = true;
builder = null; foreach (var consumer in consumers)
{
consumer?.Close();
consumer?.Dispose();
} GC.Collect();
} public static KafkaConsumer Create(string groupId, KafkaBaseOptions kafkaBaseOptions)
{
return new KafkaConsumer(groupId, kafkaBaseOptions.BootstrapServers);
}
public override string ToString()
{
return BootstrapServers;
}
} public class RecieveResult
{
CancellationTokenSource cancellationTokenSource; internal RecieveResult(ConsumeResult<string, object> consumeResult, CancellationTokenSource cancellationTokenSource)
{
this.Topic = consumeResult.Topic;
this.Message = consumeResult.Message.Value?.ToString();
this.Offset = consumeResult.Offset.Value;
this.Partition = consumeResult.Partition.Value;
this.Key = consumeResult.Message.Key; this.cancellationTokenSource = cancellationTokenSource;
} /// <summary>
/// Kafka消息所属的主题
/// </summary>
public string Topic { get; private set; }
/// <summary>
/// 键值
/// </summary>
public string Key { get; private set; }
/// <summary>
/// 我们需要处理的消息具体的内容
/// </summary>
public string Message { get; private set; }
/// <summary>
/// Kafka数据读取的当前位置
/// </summary>
public long Offset { get; private set; }
/// <summary>
/// 消息所在的物理分区
/// </summary>
public int Partition { get; private set; } /// <summary>
/// 提交
/// </summary>
public void Commit()
{
if (cancellationTokenSource == null || cancellationTokenSource.IsCancellationRequested) return; cancellationTokenSource.Cancel();
cancellationTokenSource.Dispose();
cancellationTokenSource = null;
}
}
public class ListenResult : IDisposable
{
CancellationTokenSource cancellationTokenSource; /// <summary>
/// CancellationToken
/// </summary>
public CancellationToken Token { get { return cancellationTokenSource.Token; } }
/// <summary>
/// 是否已停止
/// </summary>
public bool Stoped { get { return cancellationTokenSource.IsCancellationRequested; } } public ListenResult()
{
cancellationTokenSource = new CancellationTokenSource();
} /// <summary>
/// 停止监听
/// </summary>
public void Stop()
{
cancellationTokenSource.Cancel();
} public void Dispose()
{
Stop();
}
}

public class KafkaConverter : ISerializer<object>, IDeserializer<object>
{
/// <summary>
/// 序列化数据成字节
/// </summary>
/// <param name="data"></param>
/// <param name="context"></param>
/// <returns></returns>
public byte[] Serialize(object data, SerializationContext context)
{
var json = JsonConvert.SerializeObject(data);
return Encoding.UTF8.GetBytes(json);
}
/// <summary>
/// 反序列化字节数据成实体数据
/// </summary>
/// <param name="data"></param>
/// <param name="context"></param>
/// <returns></returns>
public object Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
{
if (isNull) return null; var json = Encoding.UTF8.GetString(data.ToArray());
try
{
return JsonConvert.DeserializeObject(json);
}
catch
{
return json;
}
}
}

public class KafkaMessage
{
/// <summary>
/// 主题
/// </summary>
public string Topic { get; set; }
/// <summary>
/// 分区，不指定分区即交给kafka指定分区
/// </summary>
public int? Partition { get; set; }
/// <summary>
/// 键值
/// </summary>
public string Key { get; set; }
/// <summary>
/// 消息
/// </summary>
public object Message { get; set; }
}

public class KafkaProducer : IDisposable
{
/// <summary>
/// 负责生成producer
/// </summary>
ProducerBuilder<string, object> builder;
ConcurrentQueue<IProducer<string, object>> producers;
bool disposed = false; /// <summary>
/// kafka服务节点
/// </summary>
public string BootstrapServers { get; private set; }
/// <summary>
/// Flush超时时间(ms)
/// </summary>
public int FlushTimeOut { get; set; } = 10000;
/// <summary>
/// 保留发布者数
/// </summary>
public int InitializeCount { get; set; } = 5;
/// <summary>
/// 默认的消息键值
/// </summary>
public string DefaultKey { get; set; }
/// <summary>
/// 默认的主题
/// </summary>
public string DefaultTopic { get; set; }
/// <summary>
/// 异常事件
/// </summary>
public event Action<object, Exception> ErrorHandler;
/// <summary>
/// 统计事件
/// </summary>
public event Action<object, string> StatisticsHandler;
/// <summary>
/// 日志事件
/// </summary>
public event Action<object, KafkaLogMessage> LogHandler; public KafkaProducer(params string[] bootstrapServers)
{
if (bootstrapServers == null || bootstrapServers.Length == 0)
{
throw new Exception("at least one server must be assigned");
} this.BootstrapServers = string.Join(",", bootstrapServers);
producers = new ConcurrentQueue<IProducer<string, object>>();
} #region Private
/// <summary>
/// producer构造器
/// </summary>
/// <returns></returns>
private ProducerBuilder<string, object> CreateProducerBuilder()
{
if (builder == null)
{
lock (this)
{
if (builder == null)
{
ProducerConfig config = new ProducerConfig();
config.BootstrapServers = BootstrapServers; //var config = new KeyValuePair<string, string>[] { new KeyValuePair<string, string>("bootstrap.servers", BootstrapServers) }; builder = new ProducerBuilder<string, object>(config);
Action<Delegate, object> tryCatchWrap = (@delegate, arg) =>
{
try
{
@delegate?.DynamicInvoke(arg);
}
catch { }
};
builder.SetErrorHandler((p, e) => tryCatchWrap(ErrorHandler, new Exception(e.Reason)));
builder.SetStatisticsHandler((p, e) => tryCatchWrap(StatisticsHandler, e));
builder.SetLogHandler((p, e) => tryCatchWrap(LogHandler, new KafkaLogMessage(e)));
builder.SetValueSerializer(new KafkaConverter());
}
}
} return builder;
}
/// <summary>
/// 租赁一个发布者
/// </summary>
/// <returns></returns>
private IProducer<string, object> RentProducer()
{
if (disposed)
{
throw new ObjectDisposedException(nameof(KafkaProducer));
} IProducer<string, object> producer;
lock (producers)
{
if (!producers.TryDequeue(out producer) || producer == null)
{
CreateProducerBuilder();
producer = builder.Build();
}
}
return producer;
}
/// <summary>
/// 返回保存发布者
/// </summary>
/// <param name="producer"></param>
private void ReturnProducer(IProducer<string, object> producer)
{
if (disposed) return; lock (producers)
{
if (producers.Count < InitializeCount && producer != null)
{
producers.Enqueue(producer);
}
else
{
producer?.Dispose();
}
}
}
#endregion #region Publish
/// <summary>
/// 发送消息
/// </summary>
/// <param name="key"></param>
/// <param name="message"></param>
/// <param name="callback"></param>
public void PublishWithKey(string key, object message, Action<DeliveryResult> callback = null)
{
Publish(DefaultTopic, null, key, message, callback);
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="partition"></param>
/// <param name="key"></param>
/// <param name="message"></param>
/// <param name="callback"></param>
public void PublishWithKey(int? partition, string key, object message, Action<DeliveryResult> callback = null)
{
Publish(DefaultTopic, partition, key, message, callback);
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="topic"></param>
/// <param name="key"></param>
/// <param name="message"></param>
/// <param name="callback"></param>
public void PublishWithKey(string topic, string key, object message, Action<DeliveryResult> callback = null)
{
Publish(topic, null, key, message, callback);
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="message"></param>
/// <param name="callback"></param>
public void Publish(object message, Action<DeliveryResult> callback = null)
{
Publish(DefaultTopic, null, DefaultKey, message, callback);
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="partition"></param>
/// <param name="message"></param>
/// <param name="callback"></param>
public void Publish(int? partition, object message, Action<DeliveryResult> callback = null)
{
Publish(DefaultTopic, partition, DefaultKey, message, callback);
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="topic"></param>
/// <param name="message"></param>
/// <param name="callback"></param>
public void Publish(string topic, object message, Action<DeliveryResult> callback = null)
{
Publish(topic, null, DefaultKey, message, callback);
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="topic"></param>
/// <param name="partition"></param>
/// <param name="message"></param>
/// <param name="callback"></param>
public void Publish(string topic, int? partition, object message, Action<DeliveryResult> callback = null)
{
Publish(topic, partition, DefaultKey, message, callback);
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="topic"></param>
/// <param name="partition"></param>
/// <param name="key"></param>
/// <param name="message"></param>
/// <param name="callback"></param>
public void Publish(string topic, int? partition, string key, object message, Action<DeliveryResult> callback = null)
{
Publish(new KafkaMessage() { Key = key, Message = message, Partition = partition, Topic = topic }, callback);
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="kafkaMessage"></param>
/// <param name="callback"></param>
public void Publish(KafkaMessage kafkaMessage, Action<DeliveryResult> callback = null)
{
if (string.IsNullOrEmpty(kafkaMessage.Topic))
{
throw new ArgumentException("topic can not be empty", nameof(kafkaMessage.Topic));
}
if (string.IsNullOrEmpty(kafkaMessage.Key))
{
throw new ArgumentException("key can not be empty", nameof(kafkaMessage.Key));
} var producer = RentProducer();
if (kafkaMessage.Partition == null)
{
producer.Produce(kafkaMessage.Topic, new Message<string, object>() { Key = kafkaMessage.Key, Value = kafkaMessage.Message }, dr => callback?.Invoke(new DeliveryResult(dr)));
}
else
{
var topicPartition = new TopicPartition(kafkaMessage.Topic, new Partition(kafkaMessage.Partition.Value));
producer.Produce(topicPartition, new Message<string, object>() { Key = kafkaMessage.Key, Value = kafkaMessage.Message }, dr => callback?.Invoke(new DeliveryResult(dr)));
} producer.Flush(TimeSpan.FromMilliseconds(FlushTimeOut)); ReturnProducer(producer);
}
#endregion #region PublishAsync
/// <summary>
/// 异步发送消息
/// </summary>
/// <param name="key"></param>
/// <param name="message"></param>
public async Task<DeliveryResult> PublishWithKeyAsync(string key, object message)
{
return await PublishAsync(DefaultTopic, null, key, message);
}
/// <summary>
/// 异步发送消息
/// </summary>
/// <param name="partition"></param>
/// <param name="key"></param>
/// <param name="message"></param>
public async Task<DeliveryResult> PublishWithKeyAsync(int? partition, string key, object message)
{
return await PublishAsync(DefaultTopic, partition, key, message);
}
/// <summary>
/// 异步发送消息
/// </summary>
/// <param name="topic"></param>
/// <param name="key"></param>
/// <param name="message"></param>
public async Task<DeliveryResult> PublishWithKeyAsync(string topic, string key, object message)
{
return await PublishAsync(topic, null, key, message);
}
/// <summary>
/// 异步发送消息
/// </summary>
/// <param name="message"></param>
public async Task<DeliveryResult> PublishAsync(object message)
{
return await PublishAsync(DefaultTopic, null, DefaultKey, message);
}
/// <summary>
/// 异步发送消息
/// </summary>
/// <param name="partition"></param>
/// <param name="message"></param>
public async Task<DeliveryResult> PublishAsync(int? partition, object message)
{
return await PublishAsync(DefaultTopic, partition, DefaultKey, message);
}
/// <summary>
/// 异步发送消息
/// </summary>
/// <param name="topic"></param>
/// <param name="message"></param>
public async Task<DeliveryResult> PublishAsync(string topic, object message)
{
return await PublishAsync(topic, null, DefaultKey, message);
}
/// <summary>
/// 异步发送消息
/// </summary>
/// <param name="topic"></param>
/// <param name="partition"></param>
/// <param name="message"></param>
public async Task<DeliveryResult> PublishAsync(string topic, int? partition, object message)
{
return await PublishAsync(topic, partition, DefaultKey, message);
}
/// <summary>
/// 异步发送消息
/// </summary>
/// <param name="topic"></param>
/// <param name="partition"></param>
/// <param name="key"></param>
/// <param name="message"></param>
/// <returns></returns>
public async Task<DeliveryResult> PublishAsync(string topic, int? partition, string key, object message)
{
return await PublishAsync(new KafkaMessage() { Key = key, Message = message, Partition = partition, Topic = topic });
}
/// <summary>
/// 异步发送消息
/// </summary>
/// <param name="kafkaMessage"></param>
/// <returns></returns>
public async Task<DeliveryResult> PublishAsync(KafkaMessage kafkaMessage)
{
if (string.IsNullOrEmpty(kafkaMessage.Topic))
{
throw new ArgumentException("topic can not be empty", nameof(kafkaMessage.Topic));
}
if (string.IsNullOrEmpty(kafkaMessage.Key))
{
throw new ArgumentException("key can not be empty", nameof(kafkaMessage.Key));
} var producer = RentProducer();
DeliveryResult<string, object> deliveryResult;
if (kafkaMessage.Partition == null)
{
deliveryResult = await producer.ProduceAsync(kafkaMessage.Topic, new Message<string, object>() { Key = kafkaMessage.Key, Value = kafkaMessage.Message });
}
else
{
var topicPartition = new TopicPartition(kafkaMessage.Topic, new Partition(kafkaMessage.Partition.Value));
deliveryResult = await producer.ProduceAsync(topicPartition, new Message<string, object>() { Key = kafkaMessage.Key, Value = kafkaMessage.Message });
} producer.Flush(new TimeSpan(0, 0, 0, 0, FlushTimeOut)); ReturnProducer(producer); return new DeliveryResult(deliveryResult);
} #endregion /// <summary>
/// 释放资源
/// </summary>
public void Dispose()
{
disposed = true;
while (producers.Count > 0)
{
IProducer<string, object> producer;
producers.TryDequeue(out producer);
producer?.Dispose();
}
GC.Collect();
} public static KafkaProducer Create(KafkaBaseOptions kafkaBaseOptions)
{
return new KafkaProducer(kafkaBaseOptions.BootstrapServers);
}
public override string ToString()
{
return BootstrapServers;
}
} public class DeliveryResult
{
internal DeliveryResult(DeliveryResult<string, object> deliveryResult)
{
this.Topic = deliveryResult.Topic;
this.Partition = deliveryResult.Partition.Value;
this.Offset = deliveryResult.Offset.Value;
switch (deliveryResult.Status)
{
case PersistenceStatus.NotPersisted: this.Status = DeliveryResultStatus.NotPersisted; break;
case PersistenceStatus.Persisted: this.Status = DeliveryResultStatus.Persisted; break;
case PersistenceStatus.PossiblyPersisted: this.Status = DeliveryResultStatus.PossiblyPersisted; break;
}
this.Key = deliveryResult.Key;
this.Message = deliveryResult.Value; if (deliveryResult is DeliveryReport<string, object>)
{
var dr = deliveryResult as DeliveryReport<string, object>;
this.IsError = dr.Error.IsError;
this.Reason = dr.Error.Reason;
}
} /// <summary>
/// 是否异常
/// </summary>
public bool IsError { get; private set; }
/// <summary>
/// 异常原因
/// </summary>
public string Reason { get; private set; }
/// <summary>
/// 主题
/// </summary>
public string Topic { get; private set; }
/// <summary>
/// 分区
/// </summary>
public int Partition { get; private set; }
/// <summary>
/// 偏移
/// </summary>
public long Offset { get; private set; }
/// <summary>
/// 状态
/// </summary>
public DeliveryResultStatus Status { get; private set; }
/// <summary>
/// 消息键值
/// </summary>
public string Key { get; private set; }
/// <summary>
/// 消息
/// </summary>
public object Message { get; private set; }
}
public enum DeliveryResultStatus
{
/// <summary>
/// 消息提交失败
/// </summary>
NotPersisted = 0,
/// <summary>
/// 消息已提交，是否成功未知
/// </summary>
PossiblyPersisted = 1,
/// <summary>
/// 消息提交成功
/// </summary>
Persisted = 2
}
public class KafkaLogMessage
{
internal KafkaLogMessage(LogMessage logMessage)
{
this.Name = logMessage.Name;
this.Facility = logMessage.Facility;
this.Message = logMessage.Message; switch (logMessage.Level)
{
case SyslogLevel.Emergency: this.Level = LogLevel.Emergency; break;
case SyslogLevel.Alert: this.Level = LogLevel.Alert; break;
case SyslogLevel.Critical: this.Level = LogLevel.Critical; break;
case SyslogLevel.Error: this.Level = LogLevel.Error; break;
case SyslogLevel.Warning: this.Level = LogLevel.Warning; break;
case SyslogLevel.Notice: this.Level = LogLevel.Notice; break;
case SyslogLevel.Info: this.Level = LogLevel.Info; break;
case SyslogLevel.Debug: this.Level = LogLevel.Debug; break;
}
}
/// <summary>
/// 名称
/// </summary>
public string Name { get; private set; }
/// <summary>
/// 级别
/// </summary>
public LogLevel Level { get; private set; }
/// <summary>
/// 装置
/// </summary>
public string Facility { get; private set; }
/// <summary>
/// 信息
/// </summary>
public string Message { get; private set; } public enum LogLevel
{
Emergency = 0,
Alert = 1,
Critical = 2,
Error = 3,
Warning = 4,
Notice = 5,
Info = 6,
Debug = 7
}
}

public class KafkaSubscriber
{
/// <summary>
/// 主题
/// </summary>
public string Topic { get; set; }
/// <summary>
/// 分区
/// </summary>
public int? Partition { get; set; }
}