/// <summary>
/// 消費者
/// </summary>
public interface IKafkaConsumer : IDisposable
{
/// <summary>
/// 消費數據
/// </summary>
/// <typeparam name="T"></typeparam>
/// <returns></returns>
T Consume<T>() where T : class;
}
public interface IKafkaProducer : IDisposable
{
/// <summary>
/// 發布消息
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="key"></param>
/// <param name="data"></param>
/// <param name="operateType"></param>
/// <returns></returns>
bool Produce<T>(string key, T data, int operateType) where T : class;
}
實現方法
using Confluent.Kafka;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Text;
namespace Kafka
{
public class KafkaConsumer : IKafkaConsumer
{
private bool disposeHasBeenCalled = false;
private readonly object disposeHasBeenCalledLockObj = new object();
private readonly IConsumer<string, string> _consumer;
/// <summary>
/// 構造函數，初始化配置
/// </summary>
/// <param name="config">配置參數</param>
/// <param name="topic">主題名稱</param>
public KafkaConsumer(ConsumerConfig config, string topic)
{
_consumer = new ConsumerBuilder<string, string>(config).Build();
_consumer.Subscribe(topic);
}
/// <summary>
/// 消費
/// </summary>
/// <returns></returns>
public T Consume<T>() where T : class
{
try
{
var result = _consumer.Consume(TimeSpan.FromSeconds(1));
if (result != null)
{
if (typeof(T) == typeof(string))
return (T)Convert.ChangeType(result.Value, typeof(T));
return JsonConvert.DeserializeObject<T>(result.Value);
}
}
catch (ConsumeException e)
{
Console.WriteLine($"consume error: {e.Error.Reason}");
}
catch (Exception e)
{
Console.WriteLine($"consume error: {e.Message}");
}
return default;
}
/// <summary>
/// 釋放
/// </summary>
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
/// <summary>
/// Dispose
/// </summary>
/// <param name="disposing"></param>
protected virtual void Dispose(bool disposing)
{
lock (disposeHasBeenCalledLockObj)
{
if (disposeHasBeenCalled) { return; }
disposeHasBeenCalled = true;
}
if (disposing)
{
_consumer?.Close();
}
}
}
}
public class KafkaProducer : IKafkaProducer
{
private bool disposeHasBeenCalled = false;
private readonly object disposeHasBeenCalledLockObj = new object();
private readonly IProducer<string, string> _producer;
private readonly string _topic;
/// <summary>
/// 構造函數，初始化配置
/// </summary>
/// <param name="config">配置參數</param>
/// <param name="topic">主題名稱</param>
public KafkaProducer(ProducerConfig config, string topic)
{
_producer = new ProducerBuilder<string, string>(config).Build();
_topic = topic;
}
/// <summary>
/// 發布消息
/// </summary>
/// <typeparam name="T">數據實體</typeparam>
/// <param name="key">數據key,partition分區會根據key</param>
/// <param name="data">數據</param>
/// <param name="operateType">操作類型[增、刪、改等不同類型]</param>
/// <returns></returns>
public bool Produce<T>(string key, T data, int operateType) where T : class
{
var obj = JsonConvert.SerializeObject(new
{
Type = operateType,
Data = data
});
try
{
var result = _producer.ProduceAsync(_topic, new Message<string, string>
{
Key = key,
Value = obj
}).ConfigureAwait(false).GetAwaiter().GetResult();
#if DEBUG
Console.WriteLine($"Topic: {result.Topic} Partition: {result.Partition} Offset: {result.Offset}");
#endif
return true;
}
catch (ProduceException<string, string> e)
{
Console.WriteLine($"Delivery failed: {e.Error.Reason}");
}
catch (Exception e)
{
Console.WriteLine($"Delivery failed: {e.Message}");
}
return false;
}
/// <summary>
/// 釋放
/// </summary>
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
/// <summary>
/// Dispose
/// </summary>
/// <param name="disposing"></param>
protected virtual void Dispose(bool disposing)
{
lock (disposeHasBeenCalledLockObj)
{
if (disposeHasBeenCalled) { return; }
disposeHasBeenCalled = true;
}
if (disposing)
{
_producer?.Dispose();
}
}
}
static void Main(string[] args)
{
var config = new ProducerConfig
{
BootstrapServers = "localhost:9092",
Acks = Acks.All
};
//發送消息
using (var kafkaProducer = new KafkaProducer(config, "topic-d"))
{
var result = kafkaProducer.Produce<object>("a", new { name = "豬八戒3" }, 1);
}
Console.WriteLine("消息發送成功");
}
static void Main(string[] args)
{
var config = new ConsumerConfig
{
BootstrapServers = "localhost:9092",
GroupId = "test",
AutoOffsetReset = AutoOffsetReset.Earliest
};
string text;
Console.WriteLine("接受中......");
while ((text = Console.ReadLine()) != "q")
{
//接受消息
using (var kafkaProducer = new KafkaConsumer(config, "topic-d"))
{
var result = kafkaProducer.Consume<object>();
if (result != null)
{
Console.WriteLine(result.ToString());
}
}
}
}
上結果、
可以看到，消息已經收到了。
歡迎各位轉載，轉載文章之后必須在文章頁面明顯位置給出作者和原文連接，否則保留追究法律責任的權利。