public class MessageService : IMessageService, ITransientDependency { private readonly IKafkaService _kafkaService; public MessageService(IKafkaService kafkaService) { _kafkaService = kafkaService; } public async Task RequestTraceAdded(XxxEventData eventData) { await _kafkaService.PublishAsync(eventData.TopicName, eventData); } }
以上相当于一个生产者，当我们消息队列发出后，还需一个消费者进行消费，所以可以使用一个控制台项目接收消息来处理业务。
var cts = new CancellationTokenSource(); Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); }; await kafkaService.SubscribeAsync<XxxEventData>(topics, async (eventData) => { // Your logic Console.WriteLine($" - {eventData.EventTime:yyyy-MM-dd HH:mm:ss} 【{eventData.TopicName}】- > 已处理"); }, cts.Token);
在 IKafkaService 中已经写了订阅消息的接口，这里也是注入后直接使用即可。
生产者消费者示例
生产者
static async Task Main(string[] args) { if (args.Length != 2) { Console.WriteLine("Usage: .. brokerList topicName"); // 127.0.0.1:9092 helloTopic return; } var brokerList = args.First(); var topicName = args.Last(); var config = new ProducerConfig { BootstrapServers = brokerList }; using var producer = new ProducerBuilder<string, string>(config).Build(); Console.WriteLine("\n-----------------------------------------------------------------------"); Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}."); Console.WriteLine("-----------------------------------------------------------------------"); Console.WriteLine("To create a kafka message with UTF-8 encoded key and value:"); Console.WriteLine("> key value<Enter>"); Console.WriteLine("To create a kafka message with a null key and UTF-8 encoded value:"); Console.WriteLine("> value<enter>"); Console.WriteLine("Ctrl-C to quit.\n"); var cancelled = false; Console.CancelKeyPress += (_, e) => { e.Cancel = true; cancelled = true; }; while (!cancelled) { Console.Write("> "); var text = string.Empty; try { text = Console.ReadLine(); } catch (IOException) { break; } if (string.IsNullOrWhiteSpace(text)) { break; } var key = string.Empty; var val = text; var index = text.IndexOf(" "); if (index != -1) { key = text.Substring(0, index); val = text.Substring(index + 1); } try { var deliveryResult = await producer.ProduceAsync(topicName, new Message<string, string> { Key = key, Value = val }); Console.WriteLine($"delivered to: {deliveryResult.TopicPartitionOffset}"); } catch (ProduceException<string, string> e) { Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}]"); } } }
消费者