345 lines
16 KiB
C#
345 lines
16 KiB
C#
|
//using Confluent.Kafka;
|
|||
|
|
//using Confluent.Kafka.Admin;
|
|||
|
|
using System;
|
|||
|
|
using System.Collections.Generic;
|
|||
|
|
using System.Linq;
|
|||
|
|
using System.Net.Http.Headers;
|
|||
|
|
using System.Runtime.CompilerServices;
|
|||
|
|
using System.Text;
|
|||
|
|
using System.Threading;
|
|||
|
|
using System.Threading.Tasks;
|
|||
|
|
|
|||
|
|
namespace APT.Infrastructure.Utility
|
|||
|
|
{
|
|||
|
|
/// <summary>
|
|||
|
|
/// kafka操作类
|
|||
|
|
/// </summary>
|
|||
|
|
//public static class KafkaHelper
|
|||
|
|
//{
|
|||
|
|
// /// <summary>
|
|||
|
|
// /// 生产
|
|||
|
|
// /// </summary>
|
|||
|
|
// /// <param name="broker">服务器ip</param>
|
|||
|
|
// /// <param name="topic">topic</param>
|
|||
|
|
// /// <param name="key">消息key</param>
|
|||
|
|
// /// <param name="text">消息内容</param>
|
|||
|
|
// /// <returns></returns>
|
|||
|
|
// public static async Task Produce(string broker, string topic, string key, string text)
|
|||
|
|
// {
|
|||
|
|
|
|||
|
|
// var config = new ProducerConfig { BootstrapServers = broker };
|
|||
|
|
|
|||
|
|
// using (var producer = new ProducerBuilder<string, string>(config).Build())
|
|||
|
|
// {
|
|||
|
|
|
|||
|
|
|
|||
|
|
// if (text == null)
|
|||
|
|
// {
|
|||
|
|
// throw (new Exception("内容不可为空"));
|
|||
|
|
|
|||
|
|
// }
|
|||
|
|
// try
|
|||
|
|
// {
|
|||
|
|
// // Note: Awaiting the asynchronous produce request below prevents flow of execution
|
|||
|
|
// // from proceeding until the acknowledgement from the broker is received (at the
|
|||
|
|
// // expense of low throughput).
|
|||
|
|
// var deliveryReport = await producer.ProduceAsync(
|
|||
|
|
// topic, new Message<string, string> { Key = key, Value = text });
|
|||
|
|
|
|||
|
|
// }
|
|||
|
|
// catch (ProduceException<string, string> e)
|
|||
|
|
// {
|
|||
|
|
// throw (new Exception($"failed to deliver message: {e.Message} [{e.Error.Code}]"));
|
|||
|
|
// }
|
|||
|
|
|
|||
|
|
// // Since we are producing synchronously, at this point there will be no messages
|
|||
|
|
// // in-flight and no delivery reports waiting to be acknowledged, so there is no
|
|||
|
|
// // need to call producer.Flush before disposing the producer.
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
|
|||
|
|
// /// <summary>
|
|||
|
|
// /// 消费
|
|||
|
|
// /// </summary>
|
|||
|
|
// /// <param name="broker">服务器ip</param>
|
|||
|
|
// /// <param name="topic">topic</param>
|
|||
|
|
// /// <param name="cts">取消令牌</param>
|
|||
|
|
// /// <param name="action">接收消息处理事件</param>
|
|||
|
|
// /// <param name="commitPeriod">每几条接收一次消息</param>
|
|||
|
|
// /// <param name="groupId">用户组</param>
|
|||
|
|
// /// <param name="isContinue">是否持续读取</param>
|
|||
|
|
// /// <returns></returns>
|
|||
|
|
// public static async Task Consume(string broker, string topic, CancellationTokenSource cts, Action<List<string>> action,
|
|||
|
|
// int commitPeriod = 5, string groupId = "optKafkaGroup", bool isContinue = false)
|
|||
|
|
// {
|
|||
|
|
// var config = new ProducerConfig { BootstrapServers = broker };
|
|||
|
|
// RunConsume(broker, new List<string> { topic }, cts.Token, commitPeriod, groupId, action, isContinue);
|
|||
|
|
// }
|
|||
|
|
|
|||
|
|
// public static async Task ConsumeParallel(string broker, string topic, CancellationTokenSource cts, Action<List<string>> action,
|
|||
|
|
// int commitPeriod = 5, string groupId = "optKafkaGroup")
|
|||
|
|
// {
|
|||
|
|
// var config = new ProducerConfig { BootstrapServers = broker };
|
|||
|
|
// RunConsumeParallel(broker, new List<string> { topic }, cts.Token, commitPeriod, groupId, action);
|
|||
|
|
// }
|
|||
|
|
|
|||
|
|
// /// <summary>
|
|||
|
|
// /// 创建消息分类
|
|||
|
|
// /// </summary>
|
|||
|
|
// /// <param name="broker">服务器</param>
|
|||
|
|
// /// <param name="topic">topic</param>
|
|||
|
|
// /// <param name="replicationFactor">备份数</param>
|
|||
|
|
// /// <param name="numPartitions">patition数</param>
|
|||
|
|
// /// <returns></returns>
|
|||
|
|
// public static async Task CreateTopicAsync(string broker, string topic, short replicationFactor = 1, short numPartitions = 1)
|
|||
|
|
// {
|
|||
|
|
// using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = broker }).Build())
|
|||
|
|
// {
|
|||
|
|
// try
|
|||
|
|
// {
|
|||
|
|
// await adminClient.CreateTopicsAsync(new TopicSpecification[] {
|
|||
|
|
// new TopicSpecification { Name = topic, ReplicationFactor = replicationFactor, NumPartitions =numPartitions } });
|
|||
|
|
// }
|
|||
|
|
// catch (CreateTopicsException e)
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine($"An error occured creating topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
|
|||
|
|
// throw e;
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
|
|||
|
|
// private static void RunConsume(string brokerList, List<string> topics, CancellationToken cancellationToken, int commitPeriod, string groupId,
|
|||
|
|
// Action<List<string>> Action, bool isContinue)
|
|||
|
|
// {
|
|||
|
|
// var config = new ConsumerConfig
|
|||
|
|
// {
|
|||
|
|
// BootstrapServers = brokerList,
|
|||
|
|
// GroupId = groupId,
|
|||
|
|
// EnableAutoCommit = false,
|
|||
|
|
// StatisticsIntervalMs = 5000,
|
|||
|
|
// SessionTimeoutMs = 6000,
|
|||
|
|
// AutoOffsetReset = AutoOffsetReset.Earliest,
|
|||
|
|
// EnablePartitionEof = true
|
|||
|
|
// };
|
|||
|
|
|
|||
|
|
// //const int commitPeriod = 5;
|
|||
|
|
|
|||
|
|
// // Note: If a key or value deserializer is not set (as is the case below), the
|
|||
|
|
// // deserializer corresponding to the appropriate type from Confluent.Kafka.Deserializers
|
|||
|
|
// // will be used automatically (where available). The default deserializer for string
|
|||
|
|
// // is UTF8. The default deserializer for Ignore returns null for all input data
|
|||
|
|
|
|||
|
|
// // (including non-null data).
|
|||
|
|
// using (var consumer = new ConsumerBuilder<Ignore, string>(config)
|
|||
|
|
// // Note: All handlers are called on the main .Consume thread.
|
|||
|
|
// .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
|
|||
|
|
// .SetStatisticsHandler((_, json) => Console.WriteLine($"Statistics: {json}"))
|
|||
|
|
// .SetPartitionsAssignedHandler((c, partitions) =>
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine($"Assigned partitions: [{string.Join(", ", partitions)}]");
|
|||
|
|
// // possibly manually specify start offsets or override the partition assignment provided by
|
|||
|
|
// // the consumer group by returning a list of topic/partition/offsets to assign to, e.g.:
|
|||
|
|
// //
|
|||
|
|
// // return partitions.Select(tp => new TopicPartitionOffset(tp, externalOffsets[tp]));
|
|||
|
|
// })
|
|||
|
|
// .SetPartitionsRevokedHandler((c, partitions) =>
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine($"Revoking assignment: [{string.Join(", ", partitions)}]");
|
|||
|
|
// })
|
|||
|
|
// .Build())
|
|||
|
|
// {
|
|||
|
|
// consumer.Subscribe(topics);
|
|||
|
|
// ConsumeResult<Ignore, string> consumeResult = null;
|
|||
|
|
// ConsumeResult<Ignore, string> lastConsumeResult = null;
|
|||
|
|
// List<string> ReciveMessage = new List<string>();
|
|||
|
|
|
|||
|
|
// try
|
|||
|
|
// {
|
|||
|
|
// while (true)
|
|||
|
|
// {
|
|||
|
|
|
|||
|
|
// try
|
|||
|
|
// {
|
|||
|
|
// consumeResult = consumer.Consume(cancellationToken);
|
|||
|
|
// if (consumeResult.IsPartitionEOF)
|
|||
|
|
// {
|
|||
|
|
// //Console.WriteLine(
|
|||
|
|
// //$"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
|
|||
|
|
// if (lastConsumeResult != null)
|
|||
|
|
// {
|
|||
|
|
// Action(ReciveMessage);//执行方法
|
|||
|
|
// try
|
|||
|
|
// {
|
|||
|
|
// consumer.Commit(lastConsumeResult);
|
|||
|
|
|
|||
|
|
// }
|
|||
|
|
// catch (KafkaException e)
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine($"Commit error: {e.Error.Reason}");
|
|||
|
|
// throw (e);
|
|||
|
|
// }
|
|||
|
|
// finally
|
|||
|
|
// {
|
|||
|
|
// ReciveMessage.Clear();
|
|||
|
|
// }
|
|||
|
|
|
|||
|
|
// }
|
|||
|
|
// if (isContinue)
|
|||
|
|
// continue;
|
|||
|
|
// else
|
|||
|
|
// break;
|
|||
|
|
// }
|
|||
|
|
// ReciveMessage.Add(consumeResult.Message.Value);
|
|||
|
|
// //Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");
|
|||
|
|
|
|||
|
|
// if (ReciveMessage.Count % commitPeriod == 0)
|
|||
|
|
// {
|
|||
|
|
// // The Commit method sends a "commit offsets" request to the Kafka
|
|||
|
|
// // cluster and synchronously waits for the response. This is very
|
|||
|
|
// // slow compared to the rate at which the consumer is capable of
|
|||
|
|
// // consuming messages. A high performance application will typically
|
|||
|
|
// // commit offsets relatively infrequently and be designed handle
|
|||
|
|
// // duplicate messages in the event of failure.
|
|||
|
|
// Action(ReciveMessage);//执行方法
|
|||
|
|
// try
|
|||
|
|
// {
|
|||
|
|
// consumer.Commit(consumeResult);
|
|||
|
|
|
|||
|
|
// }
|
|||
|
|
// catch (KafkaException e)
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine($"Commit error: {e.Error.Reason}");
|
|||
|
|
// throw (e);
|
|||
|
|
// }
|
|||
|
|
// finally
|
|||
|
|
// {
|
|||
|
|
// ReciveMessage.Clear();
|
|||
|
|
// }
|
|||
|
|
// if (isContinue)
|
|||
|
|
// continue;
|
|||
|
|
// else
|
|||
|
|
// break;
|
|||
|
|
|
|||
|
|
// }
|
|||
|
|
// lastConsumeResult = consumeResult;
|
|||
|
|
// }
|
|||
|
|
// catch (ConsumeException e)
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine($"Consume error: {e.Error.Reason}");
|
|||
|
|
// throw (e);
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
// catch (OperationCanceledException e)
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine("Closing consumer.");
|
|||
|
|
// consumer.Close();
|
|||
|
|
// throw (e);
|
|||
|
|
// }
|
|||
|
|
// catch (Exception e)
|
|||
|
|
// {
|
|||
|
|
// throw (e);
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
|
|||
|
|
// private static void RunConsumeParallel(string brokerList, List<string> topics, CancellationToken cancellationToken, int commitPeriod, string groupId,
|
|||
|
|
// Action<List<string>> Action)
|
|||
|
|
// {
|
|||
|
|
// var config = new ConsumerConfig
|
|||
|
|
// {
|
|||
|
|
// BootstrapServers = brokerList,
|
|||
|
|
// GroupId = groupId,
|
|||
|
|
// EnableAutoCommit = false,
|
|||
|
|
// StatisticsIntervalMs = 5000,
|
|||
|
|
// SessionTimeoutMs = 6000,
|
|||
|
|
// AutoOffsetReset = AutoOffsetReset.Earliest,
|
|||
|
|
// EnablePartitionEof = true
|
|||
|
|
// };
|
|||
|
|
|
|||
|
|
// //const int commitPeriod = 5;
|
|||
|
|
|
|||
|
|
// // Note: If a key or value deserializer is not set (as is the case below), the
|
|||
|
|
// // deserializer corresponding to the appropriate type from Confluent.Kafka.Deserializers
|
|||
|
|
// // will be used automatically (where available). The default deserializer for string
|
|||
|
|
// // is UTF8. The default deserializer for Ignore returns null for all input data
|
|||
|
|
|
|||
|
|
// // (including non-null data).
|
|||
|
|
// using (var consumer = new ConsumerBuilder<Ignore, string>(config)
|
|||
|
|
// // Note: All handlers are called on the main .Consume thread.
|
|||
|
|
// .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
|
|||
|
|
// .SetStatisticsHandler((_, json) => Console.WriteLine($"Statistics: {json}"))
|
|||
|
|
// .SetPartitionsAssignedHandler((c, partitions) =>
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine($"Assigned partitions: [{string.Join(", ", partitions)}]");
|
|||
|
|
// // possibly manually specify start offsets or override the partition assignment provided by
|
|||
|
|
// // the consumer group by returning a list of topic/partition/offsets to assign to, e.g.:
|
|||
|
|
// //
|
|||
|
|
// // return partitions.Select(tp => new TopicPartitionOffset(tp, externalOffsets[tp]));
|
|||
|
|
// })
|
|||
|
|
// .SetPartitionsRevokedHandler((c, partitions) =>
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine($"Revoking assignment: [{string.Join(", ", partitions)}]");
|
|||
|
|
// })
|
|||
|
|
// .Build())
|
|||
|
|
// {
|
|||
|
|
// consumer.Subscribe(topics);
|
|||
|
|
// List<ConsumeResult<Ignore, string>> consumeResultList = new List<ConsumeResult<Ignore, string>>();
|
|||
|
|
// ConsumeResult<Ignore, string> consumeResult = null;
|
|||
|
|
// ConsumeResult<Ignore, string> lastConsumeResult = null;
|
|||
|
|
// List<string> ReciveMessage = new List<string>();
|
|||
|
|
|
|||
|
|
// try
|
|||
|
|
// {
|
|||
|
|
// Task[] tasks = new Task[commitPeriod];
|
|||
|
|
// for (var x = 0; x < commitPeriod; x++)
|
|||
|
|
// {
|
|||
|
|
// tasks[x] = Task.Factory.StartNew(() =>
|
|||
|
|
// {
|
|||
|
|
// try
|
|||
|
|
// {
|
|||
|
|
// consumeResult = consumer.Consume(cancellationToken);
|
|||
|
|
// consumeResultList.Add(consumeResult);
|
|||
|
|
// if (consumeResult.IsPartitionEOF)
|
|||
|
|
// {
|
|||
|
|
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
// catch (ConsumeException e)
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine($"Consume error: {e.Error.Reason}");
|
|||
|
|
// throw (e);
|
|||
|
|
// };
|
|||
|
|
// });
|
|||
|
|
// }
|
|||
|
|
// Task.WaitAll(tasks);
|
|||
|
|
// consumeResultList = consumeResultList.OrderBy(i => i.Offset.Value).ToList();
|
|||
|
|
// foreach (var s in consumeResultList)
|
|||
|
|
// {
|
|||
|
|
// if (s.Message != null && !string.IsNullOrEmpty(s.Message.Value))
|
|||
|
|
// {
|
|||
|
|
// ReciveMessage.Add(s.Message.Value);
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
// if (ReciveMessage.Any())
|
|||
|
|
// {
|
|||
|
|
// consumer.Commit(consumeResultList.LastOrDefault());
|
|||
|
|
// Action(ReciveMessage);//执行方法
|
|||
|
|
// }
|
|||
|
|
|
|||
|
|
// }
|
|||
|
|
// catch (OperationCanceledException e)
|
|||
|
|
// {
|
|||
|
|
// Console.WriteLine("Closing consumer.");
|
|||
|
|
// consumer.Close();
|
|||
|
|
// throw (e);
|
|||
|
|
// }
|
|||
|
|
// catch (Exception e)
|
|||
|
|
// {
|
|||
|
|
// throw (e);
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
// }
|
|||
|
|
//}
|
|||
|
|
}
|