ideship/track
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor:重构预警Kafka处理，以确保服务器受支持

Browse Source
This commit is contained in:
hardstylewyl 2025-05-22 10:26:59 +08:00
parent 25eb439324
commit 584a1ca437
6 changed files with 199 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs
View File

@ -375,7 +375,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
{
var scope = _serviceProvider.CreateScope();
var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService<WarningSignalPublisher>();
_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹删除, delTracks);
_warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, delTracks);
scope.Dispose();
return;
}

4
src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs
View File

@ -522,7 +522,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
{
var scope = _serviceProvider.CreateScope();
var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService<WarningSignalPublisher>();
_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹新增, tracks);
_warningSignalPublisher.PublishSignal(WarningSignalType.轨迹新增, tracks);
scope.Dispose();
return;
}
@ -559,7 +559,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
{
var scope = _serviceProvider.CreateScope();
var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService<WarningSignalPublisher>();
_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹更新, tracks);
_warningSignalPublisher.PublishSignal(WarningSignalType.轨迹更新, tracks);
scope.Dispose();
return;
}

20
src/SJZY.Track.Application/Kafka/WarningCalcBackgroundService.cs
View File

@ -1,7 +1,5 @@
using System.Diagnostics;
using Confluent.Kafka;
using Confluent.Kafka.Core.Consumer;
using Confluent.Kafka.Core.Producer;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;
using SJZY.Expand.ABP.Core.Logger;
@ -20,23 +18,20 @@ namespace SJZY.Track.Application.Kafka;
public sealed class WarningCalcBackgroundService : BackgroundService
{
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly IKafkaConsumer<WarningSignalKey, WarningSignal> _consumer;
private readonly IKafkaProducer<WarningSignalKey, WarningSignal> _producer;
private readonly IKafkaProducer<int, WarningSignal> _dlxProducer;
private readonly IConsumer<WarningSignalKey, WarningSignal> _consumer;
private readonly WarningSignalPublisher _warningSignalPublisher;
private TrackWarningTaskOptions _options;
public WarningCalcBackgroundService(
IOptionsMonitor<TrackWarningTaskOptions> optionsMonitor,
IServiceScopeFactory serviceScopeFactory,
IKafkaConsumer<WarningSignalKey, WarningSignal> consumer,
IKafkaProducer<WarningSignalKey, WarningSignal> producer,
IKafkaProducer<int, WarningSignal> dlxProducer
IConsumer<WarningSignalKey, WarningSignal> consumer,
WarningSignalPublisher warningSignalPublisher
)
{
_serviceScopeFactory = serviceScopeFactory;
_consumer = consumer;
_producer = producer;
_dlxProducer = dlxProducer;
_warningSignalPublisher = warningSignalPublisher;
_options = optionsMonitor.CurrentValue;
optionsMonitor.OnChange(x => _options = x);
}
@ -175,7 +170,7 @@ public sealed class WarningCalcBackgroundService : BackgroundService
var tasks = needRetryMetadata.Select(async metadata =>
{
metadata.SetTrackSignalMessageHeader();
await _producer.ProduceAsync(metadata.OwenMessage, CancellationToken.None);
await _warningSignalPublisher.PublishSignalAsync(metadata.OwenMessage, CancellationToken.None);
}).ToArray();
await Task.WhenAll(tasks);
@ -193,7 +188,7 @@ public sealed class WarningCalcBackgroundService : BackgroundService
{
metadata.SetTrackSignalMessageHeader();
var msg = metadata.OwenMessage;
await _dlxProducer.ProduceAsync(new Message<int, WarningSignal>
await _warningSignalPublisher.PublishDlxSignalAsync(new Message<int, WarningSignal>
{
Key = msg.Value.OrderId,
Value = msg.Value,
@ -282,6 +277,5 @@ public sealed class WarningCalcBackgroundService : BackgroundService
base.Dispose();
_consumer.Close();
_consumer.Dispose();
_producer.Dispose();
}
}

285
src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs
View File

@ -1,20 +1,13 @@
//#define ProtoBuf
#define NewtonsoftJson
using System.Globalization;
using System.Net;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using Confluent.Kafka;
using Confluent.Kafka.Core.Consumer;
using Confluent.Kafka.Core.Producer;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Newtonsoft.Json.Converters;
using Newtonsoft.Json.Serialization;
using Microsoft.Extensions.DependencyInjection.Extensions;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Track.Application.Helpers;
using SJZY.Track.Application.Services.Warning.Signals;
using JsonSerializer = System.Text.Json.JsonSerializer;
namespace SJZY.Track.Application.Kafka;
@ -37,140 +30,180 @@ public static class WarningCalcServiceExtension
private static void RegisterConsumer<TKey, TValue>(IServiceCollection services, IConfiguration conf, string groupId, string[] subTopic)
{
services.AddSingleton<IKafkaConsumer<TKey, TValue>>(sp =>
services.AddJsonSerializer<TKey>();
services.AddJsonSerializer<TValue>();
services.AddSingleton<IConsumer<TKey, TValue>>(sp =>
{
var bb = new KafkaConsumerBuilder<TKey, TValue>()
.WithConsumerConfiguration(o =>
{
o.WithClientId(Dns.GetHostName())
.WithGroupId(groupId)
.WithBootstrapServers(conf.GetValue<string>("Kafka:BootstrapServers"))
.WithAutoOffsetReset(AutoOffsetReset.Earliest)
//偏移提交控制
.WithEnablePartitionEof(false)
.WithEnableAutoOffsetStore(false)
.WithEnableAutoCommit(false)
//链接控制
.WithSessionTimeoutMs(conf.GetValue<int>("Kafka:Consumer:SessionTimeoutMs"))
.WithMaxPollIntervalMs(conf.GetValue<int>("Kafka:Consumer:MaxPollIntervalMs"))
.WithHeartbeatIntervalMs(conf.GetValue<int>("Kafka:Consumer:HeartbeatIntervalMs"))
//吞吐量
.WithFetchMinBytes(conf.GetValue<int>("Kafka:Consumer:FetchMinBytes"))
.WithFetchMaxBytes(conf.GetValue<int>("Kafka:Consumer:FetchMaxBytes"))
.WithFetchWaitMaxMs(conf.GetValue<int>("Kafka:Consumer:FetchWaitMaxMs"))
//日志
.WithDebug("consumer,cgrp,protocol")
.WithTopicSubscriptions(subTopic);
});
bb.WithLoggerFactory(sp.GetRequiredService<ILoggerFactory>());
DebounceHelper.DebounceWrapper? reportFuncWrapper = null;
bb.WithErrorHandler((consumer, error) =>
{
if (error.IsFatal)
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"),
//日志
Debug = "consumer,cgrp,protocol",
CancellationDelayMaxMs = 100
})
.SetKeyDeserializer(sp.GetRequiredService<JsonCoreSerializer<TKey>>())
.SetValueDeserializer(sp.GetRequiredService<JsonCoreSerializer<TValue>>())
.SetErrorHandler((_, error) =>
{
LogHelper.Error($"Kafka致命错误${error}");
return;
}
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}");
// 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();
});
reportFuncWrapper.Debounce();
});
#if ProtoBuf
bb.WithProtobufNetDeserializer();
#elif NewtonsoftJson
bb.WithNewtonsoftJsonDeserializer(builder =>
builder.WithNullValueHandling(NullValueHandling.Ignore) // Ignores null values
.WithReferenceLoopHandling(ReferenceLoopHandling.Ignore) // Ignores reference loops
.WithContractResolver(new CamelCasePropertyNamesContractResolver()) // Camel case property names
);
#else
bb.WithJsonCoreDeserializer(bc =>
bc.WithReferenceHandler(ReferenceHandler.IgnoreCycles) // Handles circular references
.WithPropertyNamingPolicy(JsonNamingPolicy.CamelCase) // Uses camelCase naming
.WithDefaultIgnoreCondition(JsonIgnoreCondition.WhenWritingNull)
);
#endif
return (bb.Build() as IKafkaConsumer<TKey, TValue>)!;
var c = bb.Build();
c.Subscribe(subTopic);
return c;
});
}
private static void RegisterProducer<TKey, TValue>(IServiceCollection services, IConfiguration conf, string? defaultTopic = null)
{
services.AddSingleton<IKafkaProducer<TKey, TValue>>(sp =>
services.AddJsonSerializer<TKey>();
services.AddJsonSerializer<TValue>();
services.AddSingleton<IProducer<TKey, TValue>>(sp =>
{
var bb = new KafkaProducerBuilder<TKey, TValue>()
.WithProducerConfiguration(o =>
{
o.WithClientId(Dns.GetHostName())
.WithBootstrapServers(conf.GetValue<string>("Kafka:BootstrapServers"))
// 确保消息发送可靠
.WithAcks(Acks.All)
.WithEnableIdempotence(true) // 幂等发送
.WithMessageSendMaxRetries(conf.GetValue<int>("Kafka:Producer:MessageSendMaxRetries"))
.WithRequestTimeoutMs(conf.GetValue<int>("Kafka:Producer:RequestTimeoutMs"))
// 消息体大小
.WithCompressionType(conf.GetValue<CompressionType>("Kafka:Producer:CompressionType"))
.WithCompressionLevel(conf.GetValue<int>("Kafka:Producer:CompressionLevel")) //Gzip 0-9 LZ4 0-12 数值越大压缩效果越好cpu占用越多
// 消息发送速率控制(任意一条件先达到就发送)
.WithLingerMs(conf.GetValue<int>("Kafka:Producer:LingerMs"))
.WithBatchSize(conf.GetValue<int>("Kafka:Producer:BatchSize"))
.WithBatchNumMessages(conf.GetValue<int>("Kafka:Producer:BatchNumMessages"))
// 发送缓冲内存限制
.WithQueueBufferingMaxMessages(conf.GetValue<int>("Kafka:Producer:QueueBufferingMaxMessages"))
.WithQueueBufferingMaxKbytes(conf.GetValue<int>("Kafka:Producer:QueueBufferingMaxKbytes"))
// 分区器 一致性hash/随机
.WithPartitioner(conf.GetValue<Partitioner>("Kafka:Producer:Partitioner"))
.WithDefaultTopic(defaultTopic);
});
bb.WithLoggerFactory(sp.GetRequiredService<ILoggerFactory>());
DebounceHelper.DebounceWrapper? reportFuncWrapper = null;
bb.WithErrorHandler((producer, error) =>
{
if (error.IsFatal)
var bb = new ProducerBuilder<TKey, TValue>(new ProducerConfig()
{
LogHelper.Error($"Kafka致命错误${error}");
return;
}
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"),
})
.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}");
// 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();
});
reportFuncWrapper.Debounce();
});
#if ProtoBuf
// 2000w条消息数据包压缩从 2g -> 1g
bb.WithProtobufNetSerializer();
#elif NewtonsoftJson
bb.WithNewtonsoftJsonSerializer(builder =>
builder.WithNullValueHandling(NullValueHandling.Ignore) // Ignores null values
.WithReferenceLoopHandling(ReferenceLoopHandling.Ignore) // Ignores reference loops
.WithContractResolver(new CamelCasePropertyNamesContractResolver()) // Camel case property names
);
#else
bb.WithJsonCoreSerializer(bc =>
bc.WithReferenceHandler(ReferenceHandler.IgnoreCycles) // Handles circular references
.WithPropertyNamingPolicy(JsonNamingPolicy.CamelCase) // Uses camelCase naming
.WithDefaultIgnoreCondition(JsonIgnoreCondition.WhenWritingNull)
);
#endif
return (bb.Build() as IKafkaProducer<TKey, TValue>)!;
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;
}
}
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;
}
}

37
src/SJZY.Track.Application/Kafka/WarningSignalPublisher.cs
View File

@ -1,5 +1,5 @@
using Confluent.Kafka;
using Confluent.Kafka.Core.Producer;
using Microsoft.Extensions.Configuration;
using SJZY.Track.Application.Services.Warning.Signals;
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
using Volo.Abp.DependencyInjection;
@ -9,27 +9,48 @@ namespace SJZY.Track.Application.Kafka;
/// <summary>
/// 预警信号发布者
/// </summary>
public sealed class WarningSignalPublisher : ITransientDependency
public sealed class WarningSignalPublisher : IDisposable, ISingletonDependency
{
private readonly IKafkaProducer<WarningSignalKey, WarningSignal> _producer;
private readonly IProducer<WarningSignalKey, WarningSignal> _producer;
private readonly IProducer<int, WarningSignal> _dlxProducer;
private readonly string env;
public WarningSignalPublisher(IKafkaProducer<WarningSignalKey, WarningSignal> producer)
public WarningSignalPublisher(IProducer<WarningSignalKey, WarningSignal> producer, IProducer<int, WarningSignal> dlxProducer, IConfiguration conf)
{
_producer = producer;
_dlxProducer = dlxProducer;
env = "-" + conf.GetValue<string>("Environment")!.ToLower();
}
public void PublishWarningSignal(WarningSignalType type, List<TrackDetailsIndex> tracks)
public void PublishSignal(WarningSignalType type, List<TrackDetailsIndex> tracks)
{
foreach (var track in tracks)
PublishWarningSignal(type, track);
PublishSignal(type, track);
}
public void PublishWarningSignal(WarningSignalType type, TrackDetailsIndex track)
public void PublishSignal(WarningSignalType type, TrackDetailsIndex track)
{
_producer.Produce(new Message<WarningSignalKey, WarningSignal>
_producer.Produce(GlobalConst.WarningKafkaTopic + env, new Message<WarningSignalKey, WarningSignal>
{
Key = new WarningSignalKey(type, track.OrderId),
Value = WarningSignal.FromTrack(track, type)
});
}
public Task PublishSignalAsync(Message<WarningSignalKey, WarningSignal> msg, CancellationToken ct = default)
{
return _producer.ProduceAsync(GlobalConst.WarningKafkaTopic + env, msg, ct);
}
public Task PublishDlxSignalAsync(Message<int, WarningSignal> message, CancellationToken ct = default)
{
return _dlxProducer.ProduceAsync(GlobalConst.WarningDLXKafkaTopic + env, message, ct);
}
public void Dispose()
{
_producer.Dispose();
_dlxProducer.Dispose();
}
}

5
src/SJZY.Track.Application/SJZY.Track.Application.csproj
View File

@ -9,10 +9,7 @@
<ItemGroup>
<PackageReference Include="BSI.Stack.Rabbit" Version="2.1.4" />
<PackageReference Include="Confluent.Kafka.Core" Version="1.0.0" />
<PackageReference Include="Confluent.Kafka.Core.Serialization.JsonCore" Version="1.0.0" />
<PackageReference Include="Confluent.Kafka.Core.Serialization.NewtonsoftJson" Version="1.0.0" />
<PackageReference Include="Confluent.Kafka.Core.Serialization.ProtobufNet" Version="1.0.0" />
<PackageReference Include="Confluent.Kafka" Version="2.10.0" />
<PackageReference Include="Hangfire.AspNetCore" Version="1.8.18" />
<PackageReference Include="Hangfire.Redis.StackExchange" Version="1.9.4" />
<PackageReference Include="Savorboard.CAP.InMemoryMessageQueue" Version="8.2.1"/>