diff --git a/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs b/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs index 238f015b..91d34dec 100644 --- a/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs +++ b/src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs @@ -20,9 +20,14 @@ public interface IBroadcastingService : ITransientDependency /// /// /// - /// void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode); + /// + /// 轨迹推送完成广播（推送完成回调） + /// + /// + void TrackPushCompletedPublicTopic(List list); + /// /// 轨迹推送完成广播（推送完成回调） /// diff --git a/src/SJZY.Track.Application/Rabbit/RabbitClientBusExt.cs b/src/SJZY.Track.Application/Rabbit/RabbitClientBusExt.cs new file mode 100644 index 00000000..55ff1ae3 --- /dev/null +++ b/src/SJZY.Track.Application/Rabbit/RabbitClientBusExt.cs @@ -0,0 +1,81 @@ +using System.Diagnostics; +using BSI.Stack.Rabbit; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; +using Volo.Abp.DependencyInjection; + +namespace SJZY.Track.Application.Rabbit; + +/// +/// 支持发布确认模式 +/// +public class RabbitClientBusExt : RabbitClientBus, ISingletonDependency +{ + public RabbitClientBusExt(IServiceProvider sp, ILogger logger) + : base(sp, logger, sp.GetRequiredService>().CurrentValue) + { + } + + public void PublishByConfirmSelect( + string exchange, + string routingKey, + byte[] data, + Action? options = null + ) + { + using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey); + activity?.SetTag("kind", "publish"); + var key = exchange + ":" + routingKey; + Lazy orAdd = Channels.GetOrAdd(key, (Func>) (_ => new Lazy(CreateModel))); + var basicProperties1 = orAdd.Value.CreateBasicProperties(); + + // 配置属性 + if (options != null) + options(basicProperties1); + var model = orAdd.Value; + + // 启用发布确认 + model.ConfirmSelect(); + var readOnlyMemory = (ReadOnlyMemory) data; + model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory); + model.WaitForConfirmsOrDie(); + } + + public void PublishByConfirmSelect( + string exchange, + string routingKey, + IEnumerable datas, + Action? options = null + ) + { + using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey); + activity?.SetTag("kind", "publish"); + var key = exchange + ":" + routingKey; + Lazy orAdd = Channels.GetOrAdd(key, (Func>) (_ => new Lazy(CreateModel))); + var basicProperties1 = orAdd.Value.CreateBasicProperties(); + + // 配置属性 + if (options != null) + options(basicProperties1); + var model = orAdd.Value; + + // 启用发布确认 + model.ConfirmSelect(); + foreach (var data in datas) + { + var readOnlyMemory = (ReadOnlyMemory) data; + model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory); + } + model.WaitForConfirmsOrDie(); + } + + private IConnection GetConnection() + { + var key = Connections.OrderBy((Func, int>) (q => q.Value)).Select((Func, IConnection>) (q => q.Key)).FirstOrDefault(); + Connections.AddOrUpdate(key, (Func) (_ => 1), (Func) ((_, v) => v + 1)); + return key; + } + + private IModel CreateModel() => GetConnection().CreateModel(); +} diff --git a/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs b/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs index 8561d827..1ba4da4b 100644 --- a/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs +++ b/src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs @@ -84,7 +84,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl // 推送广播 if (logs.First().IsToMq) - logs.ForEach(log => _broadcastingService.TrackPushCompletedPublicTopic(BuildDto(log))); + _broadcastingService.TrackPushCompletedPublicTopic(logs.Select(BuildDto).ToList()); // 写入推送日志 await _trackPushRecordEsRepository.AddManyAsync(logs); diff --git a/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs index 7af7e83b..1723a226 100644 --- a/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs +++ b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs @@ -1,9 +1,9 @@ using System.Text; -using BSI.Stack.Rabbit; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; +using SJZY.Track.Application.Rabbit; namespace SJZY.Track.Application.Services.Broadcasting; @@ -12,9 +12,9 @@ namespace SJZY.Track.Application.Services.Broadcasting; /// public class BroadcastingService : IBroadcastingService { - private readonly RabbitClientBus _rabbitClientBus; + private readonly RabbitClientBusExt _rabbitClientBus; - public BroadcastingService(RabbitClientBus rabbitClientBus) + public BroadcastingService(RabbitClientBusExt rabbitClientBus) { _rabbitClientBus = rabbitClientBus; } @@ -64,15 +64,29 @@ public class BroadcastingService : IBroadcastingService else data.Option = 3; - _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}", - Encoding.UTF8.GetBytes(data.ToJson())); + _rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}", + Encoding.UTF8.GetBytes(data.ToJson()), prop => { prop.Persistent = true; }); } public void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode) { eto.Option = 2; - _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}", - Encoding.UTF8.GetBytes(eto.ToJson())); + _rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}", + Encoding.UTF8.GetBytes(eto.ToJson()), prop => { prop.Persistent = true; }); + } + + /// + /// 轨迹推送完成广播（批量） + /// + /// + /// + public void TrackPushCompletedPublicTopic(List list) + { + var datas = list + .Select(TrackPushRecordDtoToEvent) + .Select(x => Encoding.UTF8.GetBytes(x.ToJson())); + var log = list.First(); + _rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}", datas, prop => { prop.Persistent = true; }); } /// @@ -82,7 +96,24 @@ public class BroadcastingService : IBroadcastingService /// public void TrackPushCompletedPublicTopic(TrackPushRecordDto log) { - var data = new TrackPushCompletedEvent + var data = TrackPushRecordDtoToEvent(log); + _rabbitClientBus.PublishByConfirmSelect(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}", + Encoding.UTF8.GetBytes(data.ToJson()), prop => { prop.Persistent = true; }); + } + + /// + /// 轨迹推送失败广播 + /// + /// + public void TrackPushFailedPubilcTopic(TrackPushRecordDto log) + { + //暂时使用推送完成模型，后续可能会出现有区分度的模型 + TrackPushCompletedPublicTopic(log); + } + + private static TrackPushCompletedEvent TrackPushRecordDtoToEvent(TrackPushRecordDto log) + { + return new TrackPushCompletedEvent { RequestId = log.RequestId, IsSuccess = log.IsSuccess, @@ -101,14 +132,5 @@ public class BroadcastingService : IBroadcastingService TrackCode = log.TrackCode, Extra = log.Extra }; - - _rabbitClientBus.Publish(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}", - Encoding.UTF8.GetBytes(data.ToJson())); - } - - public void TrackPushFailedPubilcTopic(TrackPushRecordDto log) - { - //暂时使用推送完成模型，后续可能会出现有区分度的模型 - TrackPushCompletedPublicTopic(log); } } diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 23a5cc5f..a3020ec5 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -139,7 +139,7 @@ public class SjzyTrackApplicationModule : AbpModule Console.WriteLine("-- 当前为VS本地环境，用内存存储和内存队列"); } - context.Services.AddRabbitBoot(options => + Action action = options => { options.ConnectionString = $"amqp://{configuration["RabbitMQ:UserName"]}:{configuration["RabbitMQ:Password"]}@{configuration["RabbitMQ:HostName"]}:{configuration["RabbitMQ:Port"]}"; @@ -178,7 +178,10 @@ public class SjzyTrackApplicationModule : AbpModule } } }; - }, typeof(SjzyTrackApplicationModule).Assembly); + }; + + context.Services.Configure(action); + context.Services.AddRabbitBoot(action, typeof(SjzyTrackApplicationModule).Assembly); context.Services.Configure(configuration.GetSection("TrackApp")); context.Services.Configure(