229 lines
9.8 KiB
C#
229 lines
9.8 KiB
C#
using System.Net;
|
|
using System.Text;
|
|
using System.Text.Json;
|
|
using Confluent.Kafka;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.DependencyInjection.Extensions;
|
|
using TrackCenter.Application.Helpers;
|
|
using TrackCenter.Application.Services.Warning.Signals;
|
|
using TrackCenter.Domain.Extensions;
|
|
using JsonSerializer = System.Text.Json.JsonSerializer;
|
|
|
|
namespace TrackCenter.Application.Kafka;
|
|
|
|
public static class WarningCalcServiceExtension
|
|
{
|
|
public static void AddWarningCalcServices(this IServiceCollection services, IConfiguration conf)
|
|
{
|
|
PrintKafkaLibVersion();
|
|
var env = "-" + conf.GetValue<string>("Environment")!.ToLower();
|
|
// 预警信号批处理后台任务
|
|
services.AddHostedService<WarningCalcBackgroundService>();
|
|
|
|
// 预警信号
|
|
RegisterConsumer<WarningSignalKey, WarningSignal>(services, conf, GlobalConst.WarningKafkaGroup + env, new[] { GlobalConst.WarningKafkaTopic + env });
|
|
RegisterProducer<WarningSignalKey, WarningSignal>(services, conf, GlobalConst.WarningKafkaTopic + env);
|
|
|
|
// 预警信号DLX
|
|
RegisterConsumer<int, WarningSignal>(services, conf, GlobalConst.WarningDLXKafkaGroup + env, new[] { GlobalConst.WarningDLXKafkaTopic + env });
|
|
RegisterProducer<int, WarningSignal>(services, conf, GlobalConst.WarningDLXKafkaTopic + env);
|
|
}
|
|
|
|
private static void RegisterConsumer<TKey, TValue>(IServiceCollection services, IConfiguration conf, string groupId, string[] subTopic)
|
|
{
|
|
services.AddJsonSerializer<TKey>();
|
|
services.AddJsonSerializer<TValue>();
|
|
services.AddSingleton<IConsumer<TKey, TValue>>(sp =>
|
|
{
|
|
DebounceHelper.DebounceWrapper? reportFuncWrapper = null;
|
|
var bb = new ConsumerBuilder<TKey, TValue>(
|
|
new ConsumerConfig
|
|
{
|
|
ClientId = Dns.GetHostName(),
|
|
GroupId = groupId,
|
|
BootstrapServers = conf.GetValue<string>("Kafka:BootstrapServers"),
|
|
AutoOffsetReset = AutoOffsetReset.Earliest,
|
|
//偏移提交控制
|
|
EnablePartitionEof = false,
|
|
EnableAutoOffsetStore = false,
|
|
EnableAutoCommit = false,
|
|
//链接控制
|
|
SessionTimeoutMs = conf.GetValue<int>("Kafka:Consumer:SessionTimeoutMs"),
|
|
MaxPollIntervalMs = conf.GetValue<int>("Kafka:Consumer:MaxPollIntervalMs"),
|
|
HeartbeatIntervalMs = conf.GetValue<int>("Kafka:Consumer:HeartbeatIntervalMs"),
|
|
//吞吐量
|
|
FetchMinBytes = conf.GetValue<int>("Kafka:Consumer:FetchMinBytes"),
|
|
FetchMaxBytes = conf.GetValue<int>("Kafka:Consumer:FetchMaxBytes"),
|
|
FetchWaitMaxMs = conf.GetValue<int>("Kafka:Consumer:FetchWaitMaxMs"),
|
|
CancellationDelayMaxMs = 100,
|
|
// Debug = "all",
|
|
// LogConnectionClose = false,
|
|
// LogQueue = true
|
|
})
|
|
.SetKeyDeserializer(sp.GetRequiredService<JsonCoreSerializer<TKey>>())
|
|
.SetValueDeserializer(sp.GetRequiredService<JsonCoreSerializer<TValue>>())
|
|
.SetErrorHandler((_, error) =>
|
|
{
|
|
if (error.IsFatal)
|
|
{
|
|
LogHelper.Error($"Kafka致命错误${error}");
|
|
return;
|
|
}
|
|
|
|
// tips: 防抖以防止高频上报错误
|
|
reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () =>
|
|
{
|
|
if (error.Code == ErrorCode.Local_Transport)
|
|
LogHelper.Error($"Kafka链接错误{error}");
|
|
else if (error.Code == ErrorCode.Local_AllBrokersDown)
|
|
LogHelper.Error($"Kafka所有Broker都挂了{error}");
|
|
});
|
|
reportFuncWrapper.Debounce();
|
|
});
|
|
|
|
|
|
var c = bb.Build();
|
|
c.Subscribe(subTopic);
|
|
return c;
|
|
});
|
|
}
|
|
|
|
private static void RegisterProducer<TKey, TValue>(IServiceCollection services, IConfiguration conf, string? defaultTopic = null)
|
|
{
|
|
services.AddJsonSerializer<TKey>();
|
|
services.AddJsonSerializer<TValue>();
|
|
services.AddSingleton<IProducer<TKey, TValue>>(sp =>
|
|
{
|
|
DebounceHelper.DebounceWrapper? reportFuncWrapper = null;
|
|
var bb = new ProducerBuilder<TKey, TValue>(new ProducerConfig()
|
|
{
|
|
ClientId = Dns.GetHostName(),
|
|
BootstrapServers = conf.GetValue<string>("Kafka:BootstrapServers"),
|
|
// 确保消息发送可靠
|
|
Acks = Acks.All,
|
|
EnableIdempotence = true,
|
|
MessageSendMaxRetries = conf.GetValue<int>("Kafka:Producer:MessageSendMaxRetries"),
|
|
RequestTimeoutMs = conf.GetValue<int>("Kafka:Producer:RequestTimeoutMs"),
|
|
// 消息体大小
|
|
CompressionType = conf.GetValue<CompressionType>("Kafka:Producer:CompressionType"),
|
|
CompressionLevel = conf.GetValue<int>("Kafka:Producer:CompressionLevel"),
|
|
// 消息发送速率控制(任意一条件先达到就发送)
|
|
LingerMs = conf.GetValue<int>("Kafka:Producer:LingerMs"),
|
|
BatchSize = conf.GetValue<int>("Kafka:Producer:BatchSize"),
|
|
BatchNumMessages = conf.GetValue<int>("Kafka:Producer:BatchNumMessages"),
|
|
// 发送缓冲内存限制
|
|
QueueBufferingMaxMessages = conf.GetValue<int>("Kafka:Producer:QueueBufferingMaxMessages"),
|
|
QueueBufferingMaxKbytes = conf.GetValue<int>("Kafka:Producer:QueueBufferingMaxKbytes"),
|
|
// 分区器 一致性hash/随机
|
|
Partitioner = conf.GetValue<Partitioner>("Kafka:Producer:Partitioner"),
|
|
// Debug = "all",
|
|
// LogConnectionClose = false,
|
|
// LogQueue = true
|
|
})
|
|
.SetKeySerializer(sp.GetRequiredService<JsonCoreSerializer<TKey>>())
|
|
.SetValueSerializer(sp.GetRequiredService<JsonCoreSerializer<TValue>>())
|
|
.SetErrorHandler((_, error) =>
|
|
{
|
|
if (error.IsFatal)
|
|
{
|
|
LogHelper.Error($"Kafka致命错误${error}");
|
|
return;
|
|
}
|
|
|
|
// tips: 防抖以防止高频上报错误
|
|
reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () =>
|
|
{
|
|
if (error.Code == ErrorCode.Local_Transport)
|
|
LogHelper.Error($"Kafka链接错误{error}");
|
|
else if (error.Code == ErrorCode.Local_AllBrokersDown)
|
|
LogHelper.Error($"Kafka所有Broker都挂了{error}");
|
|
});
|
|
reportFuncWrapper.Debounce();
|
|
});
|
|
|
|
return bb.Build();
|
|
});
|
|
}
|
|
|
|
private static void AddJsonSerializer<T>(this IServiceCollection services, Action<JsonSerializerOptions>? setup = null)
|
|
{
|
|
services.TryAddSingleton(_ =>
|
|
{
|
|
var o = new JsonSerializerOptions();
|
|
setup?.Invoke(o);
|
|
return new JsonCoreSerializer<T>(o);
|
|
});
|
|
}
|
|
|
|
public static List<ConsumeResult<TKey, TValue>> ConsumeBatch<TKey, TValue>(this IConsumer<TKey, TValue> consumer, int batchSize, TimeSpan timeout)
|
|
{
|
|
using var cts = new CancellationTokenSource(timeout);
|
|
var list = new List<ConsumeResult<TKey, TValue>>();
|
|
while (list.Count < batchSize && !cts.IsCancellationRequested)
|
|
{
|
|
var consumeResult = consumer.Consume(100);
|
|
if (consumeResult != null)
|
|
list.Add(consumeResult);
|
|
}
|
|
|
|
return list;
|
|
}
|
|
|
|
public static void PrintKafkaLibVersion()
|
|
{
|
|
var loadSuccess = Library.Load(null);
|
|
Console.WriteLine($@"
|
|
+-------------------------------------------------------+
|
|
| |
|
|
| LibrdKafka |
|
|
| 加载：{loadSuccess}，版本号：{Library.VersionString} |
|
|
| |
|
|
+-------------------------------------------------------+
|
|
");
|
|
|
|
}
|
|
}
|
|
|
|
internal sealed class JsonCoreSerializer<T> : ISerializer<T>, IDeserializer<T>
|
|
{
|
|
private readonly Encoding _encoding;
|
|
private readonly JsonSerializerOptions _options;
|
|
|
|
public JsonCoreSerializer(JsonSerializerOptions options)
|
|
{
|
|
_options = options ?? throw new ArgumentNullException(nameof(options));
|
|
_encoding = Encoding.UTF8;
|
|
}
|
|
|
|
public byte[] Serialize(T data, SerializationContext context)
|
|
{
|
|
if (data is null)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
var json = JsonSerializer.Serialize(data, data.GetType(), _options);
|
|
|
|
var result = _encoding.GetBytes(json);
|
|
|
|
return result;
|
|
}
|
|
|
|
public T Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
|
|
{
|
|
if (isNull)
|
|
{
|
|
return default;
|
|
}
|
|
|
|
#if NETSTANDARD2_0
|
|
var json = _encoding.GetString(data.ToArray());
|
|
#else
|
|
var json = _encoding.GetString(data);
|
|
#endif
|
|
var result = JsonSerializer.Deserialize<T>(json, _options);
|
|
|
|
return result;
|
|
}
|
|
}
|