diff --git a/Directory.Build.props b/Directory.Build.props index 1f5eaf03..a4b16976 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -4,6 +4,6 @@ - 8.0.6 + 8.0.6.2 diff --git a/src/TrackCenter.Api/TrackCenter.Api.csproj b/src/TrackCenter.Api/TrackCenter.Api.csproj index 48e85830..9dff28a6 100644 --- a/src/TrackCenter.Api/TrackCenter.Api.csproj +++ b/src/TrackCenter.Api/TrackCenter.Api.csproj @@ -20,8 +20,7 @@ - - + diff --git a/src/TrackCenter.Api/TrackCenterApiModule.cs b/src/TrackCenter.Api/TrackCenterApiModule.cs index 7af17e9d..6fbba35d 100644 --- a/src/TrackCenter.Api/TrackCenterApiModule.cs +++ b/src/TrackCenter.Api/TrackCenterApiModule.cs @@ -5,8 +5,8 @@ using Fake.AspNetCore.ExceptionHandling; using Fake.AspNetCore.Newtonsoft; using Fake.AspNetCore.Swagger; using Fake.Autofac; -using Fake.EventBus.Dashboard; -using Fake.EventBus.IntegrationEventLog; +using Fake.Messaging.Dashboard; +using Fake.Messaging.EntityFrameworkCore; using Fake.FeiShu; using Fake.Json; using Fake.Modularity; @@ -33,9 +33,8 @@ namespace TrackCenter.Api; [DependsOn(typeof(FakeAspNetCoreModule))] [DependsOn(typeof(TrackCenterApplicationModule))] [DependsOn(typeof(TrackCenterRepositoryModule))] -// RabbitMQ 物理总线之后 Replace IDistributedEventBus → Outbox（与 admin 一致） -[DependsOn(typeof(FakeEventBusIntegrationEventLogModule))] -[DependsOn(typeof(FakeEventBusDashboardModule))] +// Messaging 独立队列（Outbox/Inbox + Dashboard）；物理传输由 FakeMessagingRabbitMqModule 提供 +[DependsOn(typeof(FakeMessagingDashboardModule))] [DependsOn(typeof(FakeSerilogModule))] public class TrackCenterApiModule : FakeModule { @@ -59,10 +58,11 @@ public class TrackCenterApiModule : FakeModule // 配置签名验证选项 services.Configure(configuration.GetSection("SignatureValidation")); - // Outbox / Inbox 与业务同库（EF），通过 DbTransaction 征用 SqlSugar 事务 - services.AddDbContext(options => + // Outbox / Inbox（fk_msg_*）与业务同库（EF），通过 DbTransaction 征用 SqlSugar 事务 + services.AddDbContext(options => { - var connectionString = configuration.GetConnectionString("IntegrationEventLog") + var connectionString = configuration.GetConnectionString("Messaging") + ?? configuration.GetConnectionString("IntegrationEventLog") ?? configuration.GetConnectionString("Default") ?? configuration["ConnectionStrings:Default"] ?? string.Empty; @@ -73,6 +73,13 @@ public class TrackCenterApiModule : FakeModule #endif }); + // Dashboard 路径保持 /eventbus，兼容旧书签与网关 + services.Configure(o => + { + if (string.IsNullOrWhiteSpace(o.PathMatch) || o.PathMatch == "/messaging") + o.PathMatch = "/eventbus"; + }); + // 全局授权配置：所有接口默认需要授权 services.AddMvc(options => { @@ -207,7 +214,7 @@ public class TrackCenterApiModule : FakeModule }); // 必须在 UsePathBase 之后：完整对外路径为 {PathBase}/eventbus - app.UseFakeEventBusDashboard(); + app.UseFakeMessagingDashboard(); app.Services.GetRequiredService().Send("Application started"); } diff --git a/src/TrackCenter.Application/Rabbit/InterfaceRelayForwardReplayEvent.cs b/src/TrackCenter.Application/Events/InterfaceRelayForwardReplayEvent.cs similarity index 70% rename from src/TrackCenter.Application/Rabbit/InterfaceRelayForwardReplayEvent.cs rename to src/TrackCenter.Application/Events/InterfaceRelayForwardReplayEvent.cs index 0f9144ba..c3b171ef 100644 --- a/src/TrackCenter.Application/Rabbit/InterfaceRelayForwardReplayEvent.cs +++ b/src/TrackCenter.Application/Events/InterfaceRelayForwardReplayEvent.cs @@ -1,6 +1,12 @@ -namespace TrackCenter.Application.Rabbit; +using Fake.Messaging; -public class InterfaceRelayForwardReplayEvent +namespace TrackCenter.Application.Events; + +/// +/// 接口通回调报文（外部生产，无 Fake MessageId / DLX）。 +/// +[MessageRoute(Topic = GlobalConst.InterfaceRelayExchange, TopicType = "direct")] +public class InterfaceRelayForwardReplayEvent : Message { /// /// 请求Id diff --git a/src/TrackCenter.Application/Events/OrderPodPullEvent.cs b/src/TrackCenter.Application/Events/OrderPodPullEvent.cs index 3d09e256..e30746b0 100644 --- a/src/TrackCenter.Application/Events/OrderPodPullEvent.cs +++ b/src/TrackCenter.Application/Events/OrderPodPullEvent.cs @@ -1,4 +1,4 @@ -using Fake.EventBus.Distributed; +using Fake.Messaging; namespace TrackCenter.Application.Events; @@ -6,7 +6,8 @@ namespace TrackCenter.Application.Events; /// 订单 POD 拉取（当前仓库无订阅方，仍走 Outbox 投递以兼容后续消费者）。 /// [Serializable] -public class OrderPodPullEvent : IntegrationEvent +[MessageRoute(nameof(OrderPodPullEvent))] +public class OrderPodPullEvent : Message { // 确定应用 public string AppKey { get; set; } diff --git a/src/TrackCenter.Application/Events/TrackChangeAfterUpdateOrderEvent.cs b/src/TrackCenter.Application/Events/TrackChangeAfterUpdateOrderEvent.cs index e438598a..2f2acd1b 100644 --- a/src/TrackCenter.Application/Events/TrackChangeAfterUpdateOrderEvent.cs +++ b/src/TrackCenter.Application/Events/TrackChangeAfterUpdateOrderEvent.cs @@ -1,9 +1,10 @@ -using Fake.EventBus.Distributed; +using Fake.Messaging; namespace TrackCenter.Application.Events; [Serializable] -public class TrackChangeAfterUpdateOrderEvent : IntegrationEvent +[MessageRoute(nameof(TrackChangeAfterUpdateOrderEvent))] +public class TrackChangeAfterUpdateOrderEvent : Message { /// /// 订单Id diff --git a/src/TrackCenter.Application/Events/TrackDeleteEvent.cs b/src/TrackCenter.Application/Events/TrackDeleteEvent.cs index 8cac8130..46342712 100644 --- a/src/TrackCenter.Application/Events/TrackDeleteEvent.cs +++ b/src/TrackCenter.Application/Events/TrackDeleteEvent.cs @@ -1,10 +1,11 @@ -using Fake.EventBus.Distributed; +using Fake.Messaging; using TrackCenter.Application.Dtos.ProvideServices; namespace TrackCenter.Application.Events; [Serializable] -public class TrackDeleteEvent : IntegrationEvent +[MessageRoute(nameof(TrackDeleteEvent))] +public class TrackDeleteEvent : Message { /// /// 根据esId定位轨迹，优先级最高 diff --git a/src/TrackCenter.Application/Events/TrackPushEvent.cs b/src/TrackCenter.Application/Events/TrackPushEvent.cs index 53a63a8e..bd481ed1 100644 --- a/src/TrackCenter.Application/Events/TrackPushEvent.cs +++ b/src/TrackCenter.Application/Events/TrackPushEvent.cs @@ -1,4 +1,4 @@ -using Fake.EventBus.Distributed; +using Fake.Messaging; namespace TrackCenter.Application.Events; @@ -7,7 +7,8 @@ namespace TrackCenter.Application.Events; /// 每次可传递多个订单，但仅能传递同一个轨迹内容 /// [Serializable] -public class TrackPushEvent : IntegrationEvent +[MessageRoute(nameof(TrackPushEvent))] +public class TrackPushEvent : Message { /// /// 操作类型 diff --git a/src/TrackCenter.Application/Events/TrackUpsertEvent.cs b/src/TrackCenter.Application/Events/TrackUpsertEvent.cs index ee4ac79f..1931d689 100644 --- a/src/TrackCenter.Application/Events/TrackUpsertEvent.cs +++ b/src/TrackCenter.Application/Events/TrackUpsertEvent.cs @@ -1,10 +1,11 @@ -using Fake.EventBus.Distributed; +using Fake.Messaging; using TrackCenter.Application.Dtos.ProvideServices; namespace TrackCenter.Application.Events; [Serializable] -public class TrackUpsertEvent : IntegrationEvent +[MessageRoute(nameof(TrackUpsertEvent))] +public class TrackUpsertEvent : Message { /// /// 应用 diff --git a/src/TrackCenter.Application/Events/TrackUpsertForRefreshEvent.cs b/src/TrackCenter.Application/Events/TrackUpsertForRefreshEvent.cs index ef946ab7..5aefc531 100644 --- a/src/TrackCenter.Application/Events/TrackUpsertForRefreshEvent.cs +++ b/src/TrackCenter.Application/Events/TrackUpsertForRefreshEvent.cs @@ -1,9 +1,11 @@ +using Fake.Messaging; namespace TrackCenter.Application.Events; /// /// 轨迹刷新引发的新增/更新（独立队列，与 隔离）。 /// [Serializable] +[MessageRoute(nameof(TrackUpsertForRefreshEvent))] public class TrackUpsertForRefreshEvent : TrackUpsertEvent { } diff --git a/src/TrackCenter.Application/Messaging/TrackChangeAfterUpdateOrderEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackChangeAfterUpdateOrderEventHandler.cs index 0da20f58..545011fe 100644 --- a/src/TrackCenter.Application/Messaging/TrackChangeAfterUpdateOrderEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackChangeAfterUpdateOrderEventHandler.cs @@ -1,6 +1,4 @@ -using Fake.EventBus; -using Fake.EventBus.Distributed; -using Fake.UnitOfWork; +using Fake.Messaging; using TrackCenter.Application.Events; using TrackCenter.Application.Services.Warning.Handlers; using TrackCenter.Domain.Aggregates.OrderAggregate; @@ -8,7 +6,7 @@ using TrackCenter.Domain.Aggregates.OrderAggregate; namespace TrackCenter.Application.Messaging; public class TrackChangeAfterUpdateOrderEventHandler(IOrderRepository orderRepository, IServiceProvider serviceProvider) - : IEventHandler, ITransientDependency + : IMessageHandle, ITransientDependency { public virtual async Task HandleAsync( TrackChangeAfterUpdateOrderEvent @event, diff --git a/src/TrackCenter.Application/Messaging/TrackDeleteEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackDeleteEventHandler.cs index 09105b19..da2e0850 100644 --- a/src/TrackCenter.Application/Messaging/TrackDeleteEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackDeleteEventHandler.cs @@ -1,5 +1,4 @@ -using Fake.EventBus; -using Fake.EventBus.Distributed; +using Fake.Messaging; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Dtos.ProvideServices; @@ -17,23 +16,23 @@ namespace TrackCenter.Application.Messaging; public class TrackDeleteEventHandler( ProvideQueries provideQueries, IOrderRepository orderRepository, - IDistributedEventBus eventBus, + IMessagePublisher messagePublisher, ITrackDetailsEsRepository trackDetailsEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, BroadcastingService broadcastingService, WarningSignalPublisher warningSignalPublisher) - : IEventHandler, ITransientDependency + : IMessageHandle, ITransientDependency { [UnitOfWork(IsTransactional = true)] - public virtual async Task HandleAsync(TrackDeleteEvent @event, CancellationToken cancellationToken = default) + public virtual async Task HandleAsync(TrackDeleteEvent message, CancellationToken cancellationToken = default) { - if (!CanDelete(@event)) return; + if (!CanDelete(message)) return; // 处理删除轨迹 - await HandleDeleteTrackAsync(@event); + await HandleDeleteTrackAsync(message); // 根据订单关联关系异步向下刷（异步） - await DownByOrderLink(@event); + await DownByOrderLink(message); } private async Task DownByOrderLink(TrackDeleteEvent @event) @@ -77,7 +76,7 @@ public class TrackDeleteEventHandler( var subEvent = nextEvent.DeepCopy(); subEvent.OrderIds = subOrderIds; // 关联单继续走删除链路（原 CAP 误发到 Upsert topic，此处修正） - await eventBus.PublishAsync(subEvent); + await messagePublisher.PublishAsync(subEvent); } } } diff --git a/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackPushCompletedEventHandler.cs similarity index 65% rename from src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs rename to src/TrackCenter.Application/Messaging/TrackPushCompletedEventHandler.cs index 5c1ed968..8567f549 100644 --- a/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackPushCompletedEventHandler.cs @@ -1,63 +1,44 @@ -using System.Text; +using Fake.Messaging; using Fake.UnitOfWork; +using InterfaceForward.Domain.Shared.Helpers; using Newtonsoft.Json; -using RabbitMQ.Client.Events; using TrackCenter.Application.Dtos.TrackPush; +using TrackCenter.Application.Events; using TrackCenter.Application.Helpers; using TrackCenter.Application.Services.Broadcasting; using TrackCenter.Application.Services.Warning.Handlers; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackPushAggregate; -namespace TrackCenter.Application.Rabbit; +namespace TrackCenter.Application.Messaging; -[ExposeServices(typeof(IRabbitHandler))] -public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandler +/// +/// 接口通轨迹推送完成回调 → 独立队列 。 +/// +[MessageSubscribe(GlobalConst.TrackPushCompletedSubscribe, ExternalQueue = true)] +public class TrackPushCompletedEventHandler( + IOrderRepository orderRepository, + ITrackPushRecordEsRepository trackPushRecordEsRepository, + BroadcastingService broadcastingService, + IReturnWarningHandler returnWarningHandler) + : IMessageHandle, ITransientDependency { - private readonly IOrderRepository _orderRepository; - private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; - private readonly BroadcastingService _broadcastingService; - private readonly IReturnWarningHandler _returnWarningHandler; - - public TrackPushCompletedEventHandler(IOrderRepository orderRepository, - ITrackPushRecordEsRepository trackPushRecordEsRepository, - BroadcastingService broadcastingService, IReturnWarningHandler returnWarningHandler) + public virtual async Task HandleAsync( + InterfaceRelayForwardReplayEvent message, + CancellationToken cancellationToken = default) { - _orderRepository = orderRepository; - _trackPushRecordEsRepository = trackPushRecordEsRepository; - _broadcastingService = broadcastingService; - _returnWarningHandler = returnWarningHandler; - } - - public bool Enable(ConsumeOptions options) - { - return options.Queue.StartsWith(GlobalConst.TrackPushCompletedSubscribe); - } - - [UnitOfWork(IsTransactional = true)] - public virtual async Task Handle(BasicDeliverEventArgs args) - { - var msg = string.Empty; try { - msg = Encoding.UTF8.GetString(args.Body.ToArray()); - var @event = System.Text.Json.JsonSerializer.Deserialize(msg); - if (@event == null) - { - Log.Warning( - $"rabbit on queue received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); - return; - } - - await ConsumerAsync(@event); + await ConsumerAsync(message); } catch (Exception ex) { - Log.Error($"TrackPushCompletedEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{msg}"); + Log.Error($"TrackPushCompletedEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{message.ToJson()}"); + throw; } } - public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) + private async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) { var optionType = Base64Helper.DeSerialize(@event.Attach); var pushInputs = new List(); @@ -91,16 +72,16 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl // 推送广播 if (logs.First().IsToMq) - _broadcastingService.TrackPushCompletedPublicTopic(logs.Select(log => BuildDto(log, @event.Data)).ToList()); + broadcastingService.TrackPushCompletedPublicTopic(logs.Select(log => BuildDto(log, @event.Data)).ToList()); // 写入推送日志 - await _trackPushRecordEsRepository.AddManyAsync(logs); + await trackPushRecordEsRepository.AddManyAsync(logs); // 同步更新订单主表 await TrackPushAfterUpdateOrderAsync(isAdd, @event.IsSuccess, logs); // 引发预警计算 - await _returnWarningHandler.ReceiveTrackPushRecordHandleAsync(@event.IsSuccess, @event.RequestParams, logs, pushInputs, updateInputs); + await returnWarningHandler.ReceiveTrackPushRecordHandleAsync(@event.IsSuccess, @event.RequestParams, logs, pushInputs, updateInputs); } private async Task TrackPushAfterUpdateOrderAsync(bool isAdd, bool isSuccess, List logs) @@ -118,7 +99,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl LatestPushError = x.Message }).ToList(); - await _orderRepository.UpdateRangeAsync(orders, x => new + await orderRepository.UpdateRangeAsync(orders, x => new { x.QuestionType, x.LatestPushError @@ -136,7 +117,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl LatestPushTime = x.CreateTime, }).ToList(); - await _orderRepository.UpdateRangeAsync(orders, x => new + await orderRepository.UpdateRangeAsync(orders, x => new { x.QuestionType, x.LatestPushError, diff --git a/src/TrackCenter.Application/Messaging/TrackPushEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackPushEventHandler.cs index 206796f1..8f852117 100644 --- a/src/TrackCenter.Application/Messaging/TrackPushEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackPushEventHandler.cs @@ -1,7 +1,6 @@ using System.Collections.Immutable; using System.Text.Json; -using Fake.EventBus; -using Fake.EventBus.Distributed; +using Fake.Messaging; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Options; @@ -29,7 +28,7 @@ public class TrackPushEventHandler( BroadcastingService broadcastingService, ITrackDetailsEsRepository trackDetailsEsRepository, InterfaceForwardService interfaceForwardService) - : IEventHandler, ITransientDependency + : IMessageHandle, ITransientDependency { private List _orders = null!; private List _pushConfigs = null!; diff --git a/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackRefreshEventHandler.cs similarity index 91% rename from src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs rename to src/TrackCenter.Application/Messaging/TrackRefreshEventHandler.cs index ade732fe..544e4251 100644 --- a/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackRefreshEventHandler.cs @@ -1,13 +1,11 @@ using System.Diagnostics; -using System.Text; -using Fake.EventBus.Distributed; +using Fake.Messaging; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Newtonsoft.Json; using Newtonsoft.Json.Linq; -using RabbitMQ.Client.Events; using TrackCenter.Application.Dtos.TrackPull; using TrackCenter.Application.Events; using TrackCenter.Application.Helpers; @@ -16,47 +14,37 @@ using TrackCenter.Application.Services.TrackPull; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; -namespace TrackCenter.Application.Rabbit; +namespace TrackCenter.Application.Messaging; -[ExposeServices(typeof(IRabbitHandler))] +/// +/// 接口通轨迹刷新回调 → 独立队列 。 +/// +[MessageSubscribe(GlobalConst.TrackRefreshSubscribe, ExternalQueue = true)] public class TrackRefreshEventHandler( IOrderRepository orderRepository, - IDistributedEventBus eventBus, + IMessagePublisher messagePublisher, TrackPullCacheService trackPullCacheService, ITrackDetailsEsRepository trackDetailsEsRepository, ILogger logger, IOptionsMonitor options) - : ITransientDependency, IRabbitHandler + : IMessageHandle, ITransientDependency { - public bool Enable(ConsumeOptions coptions) + public virtual async Task HandleAsync( + InterfaceRelayForwardReplayEvent message, + CancellationToken cancellationToken = default) { - return coptions.Queue.StartsWith(GlobalConst.TrackRefreshSubscribe); - } - - [UnitOfWork(IsTransactional = true)] - public virtual async Task Handle(BasicDeliverEventArgs args) - { - var msg = string.Empty; try { - msg = Encoding.UTF8.GetString(args.Body.ToArray()); - var @event = System.Text.Json.JsonSerializer.Deserialize(msg); - if (@event == null) - { - logger.LogWarning( - $"rabbit on queue received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); - return; - } - - await ConsumerAsync(@event); + await ConsumerAsync(message); } catch (Exception ex) { - Log.Error($"TrackRefreshEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{msg}"); + Log.Error($"TrackRefreshEventHandler出错，错误信息:{ex}\r\n-----\r\n消息:{message.ToJson()}"); + throw; } } - public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) + private async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) { if (!@event.IsSuccess) { @@ -71,15 +59,6 @@ public class TrackRefreshEventHandler( return; } - // 最优先级，最匹配的 权重 sort - - // 客户、渠道、产品 - // */*/* 全部 - // 客户A/渠道A/产品A - // 客户A - - // 客户A 渠道B 产品A - var config = await trackPullCacheService.GetPullParamsWithContentAsync(attach.AppKey, attach.ServiceProviderCode, attach.OrderType); if (config is null) @@ -134,7 +113,6 @@ public class TrackRefreshEventHandler( ServiceProviderCode = attach.ServiceProviderCode }; - if (trackToken is JArray trackTokens) { foreach (var item in trackTokens) @@ -284,7 +262,7 @@ public class TrackRefreshEventHandler( Extra = trackExtra }; - await eventBus.PublishAsync(trackUpsertEvent); + await messagePublisher.PublishAsync(trackUpsertEvent); } private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace, diff --git a/src/TrackCenter.Application/Messaging/TrackUpsertEventHandler.cs b/src/TrackCenter.Application/Messaging/TrackUpsertEventHandler.cs index 461d37eb..c464502f 100644 --- a/src/TrackCenter.Application/Messaging/TrackUpsertEventHandler.cs +++ b/src/TrackCenter.Application/Messaging/TrackUpsertEventHandler.cs @@ -1,5 +1,4 @@ -using Fake.EventBus; -using Fake.EventBus.Distributed; +using Fake.Messaging; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Events; @@ -18,30 +17,22 @@ namespace TrackCenter.Application.Messaging; public class TrackUpsertEventHandler( IOrderRepository orderRepository, - IDistributedEventBus eventBus, + IMessagePublisher messagePublisher, ProvideQueries provideQueries, ITrackDetailsEsRepository trackDetailsEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, BroadcastingService broadcastingService, WarningSignalPublisher warningSignalPublisher) : - IEventHandler, - IEventHandler, + IMessageHandle, + IMessageHandle, ITransientDependency { - Task IEventHandler.HandleAsync(Event eventData, CancellationToken cancellationToken) - => eventData switch - { - TrackUpsertForRefreshEvent refresh => HandleAsync(refresh, cancellationToken), - TrackUpsertEvent upsert => HandleAsync(upsert, cancellationToken), - _ => Task.CompletedTask - }; + public Task HandleAsync(TrackUpsertEvent message, CancellationToken cancellationToken = default) + => ConsumerAsync(message); - public Task HandleAsync(TrackUpsertEvent @event, CancellationToken cancellationToken = default) - => ConsumerAsync(@event); - - public Task HandleAsync(TrackUpsertForRefreshEvent @event, CancellationToken cancellationToken = default) - => ConsumerAsync(@event); + public Task HandleAsync(TrackUpsertForRefreshEvent message, CancellationToken cancellationToken = default) + => ConsumerAsync(message); private async Task ConsumerAsync(TrackUpsertEvent @event) { @@ -107,7 +98,7 @@ public class TrackUpsertEventHandler( { var subEvent = nextEvent.DeepCopy(); subEvent.OrderIdentifiers = subOrderIds.ToArray(); - await eventBus.PublishAsync(subEvent); + await messagePublisher.PublishAsync(subEvent); } } } @@ -200,7 +191,7 @@ public class TrackUpsertEventHandler( // 如果存在轨迹标记，则触发订单pod拉取 if (!string.IsNullOrWhiteSpace(trackContent.TrackFlagId)) { - await eventBus.PublishAsync(new OrderPodPullEvent + await messagePublisher.PublishAsync(new OrderPodPullEvent { AppKey = @event.AppKey, OrderIdEsIdDict = insertTracks.ToDictionary(x => x.OrderId, y => y.Id), @@ -213,7 +204,7 @@ public class TrackUpsertEventHandler( await TrackChangePublicTopicAsync(trackContent, trackChangeEtos, true); // 触发订单主表更新事件（异步） - await eventBus.PublishAsync(new TrackChangeAfterUpdateOrderEvent + await messagePublisher.PublishAsync(new TrackChangeAfterUpdateOrderEvent { OrderIds = canNextOrderIds.ToArray(), IsCompleted = trackContent.IsCompleted, @@ -248,7 +239,7 @@ public class TrackUpsertEventHandler( ForcedPush = @event.ForcedPush, IsToMq = (trackContent.IsToMq || @event.IsToMq) }; - await eventBus.PublishAsync(trackPushEvent); + await messagePublisher.PublishAsync(trackPushEvent); } private async Task> UpsertTrackAsync( diff --git a/src/TrackCenter.Application/Rabbit/IRabbitHandler.cs b/src/TrackCenter.Application/Rabbit/IRabbitHandler.cs deleted file mode 100644 index 19635f18..00000000 --- a/src/TrackCenter.Application/Rabbit/IRabbitHandler.cs +++ /dev/null @@ -1,10 +0,0 @@ -using RabbitMQ.Client.Events; - -namespace TrackCenter.Application.Rabbit; - -public interface IRabbitHandler -{ - bool Enable(ConsumeOptions options); - - Task Handle(BasicDeliverEventArgs args); -} diff --git a/src/TrackCenter.Application/Rabbit/RabbitClient.cs b/src/TrackCenter.Application/Rabbit/RabbitClient.cs deleted file mode 100644 index ac0ebdae..00000000 --- a/src/TrackCenter.Application/Rabbit/RabbitClient.cs +++ /dev/null @@ -1,166 +0,0 @@ -using Fake.RabbitMQ; -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; -using RabbitMQ.Client; -using RabbitMQ.Client.Events; - -namespace TrackCenter.Application.Rabbit; - -public class RabbitClient( - IRabbitMqChannelPool channelPool, - IServiceScopeFactory serviceScopeFactory, - ILogger logger, - IOptions options) : ISingletonDependency, IHostedService -{ - public void Subscribe(ConsumeOptions consumeOptions) - { - using var channelAccessor = channelPool.Acquire(consumeOptions.Queue); - - if (!channelAccessor.IsNew) return; - var channel = channelAccessor.Channel; - var decl = new DefaultDeclaration(channel); - - consumeOptions.Declaration.Invoke(decl); - - if (consumeOptions.FetchCount > 0) - { - channel.BasicQos(0, consumeOptions.FetchCount, false); - } - - var consumer = new AsyncEventingBasicConsumer(channel); - consumer.Received += (model, ea) => - { - _ = Task.Run(async () => - { - try - { - await using var scope = serviceScopeFactory.CreateAsyncScope(); - foreach (var handler in scope.ServiceProvider.GetServices()) - { - if (handler.Enable(consumeOptions)) - { - await handler.Handle(ea); - } - } - channel.BasicAck(ea.DeliveryTag, false); - } - catch (Exception ex) - { - logger.LogError($"rabbit on queue({consumeOptions.Queue}) received error: {ex}"); - channel.BasicNack(ea.DeliveryTag, false, consumeOptions.FailedRequeue); - } - }); - - return Task.CompletedTask; - }; - - channel.CallbackException += (sender, args) => - { - logger.LogError(args.Exception, "RabbitMQ channel exception"); - }; - - channel.BasicConsume(queue: consumeOptions.Queue, - autoAck: false, - consumer: consumer); - } - - public void Publish(string exchange, string routingKey, byte[] data, Action? options = null) - { - using var channelAccessor = channelPool.Acquire(routingKey); - var props = channelAccessor.Channel.CreateBasicProperties(); - options?.Invoke(props); - channelAccessor.Channel.BasicPublish(exchange, routingKey, props, data); - } - - public void PublishByConfirmSelect( - string exchange, - string routingKey, - byte[] data, - Action? options = null - ) - { - using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey); - activity?.SetTag("kind", "publish"); - using var channelAccessor = channelPool.Acquire(routingKey); - var basicProperties = channelAccessor.Channel.CreateBasicProperties(); - - // 配置属性 - if (options != null) - options(basicProperties); - var model = channelAccessor.Channel; - - // 启用发布确认 - model.ConfirmSelect(); - var readOnlyMemory = (ReadOnlyMemory) data; - model.BasicPublish(exchange, routingKey, basicProperties, 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"); - using var channelAccessor = channelPool.Acquire(routingKey); - var basicProperties = channelAccessor.Channel.CreateBasicProperties(); - - // 配置属性 - if (options != null) - options(basicProperties); - var model = channelAccessor.Channel; - - // 启用发布确认 - model.ConfirmSelect(); - foreach (var data in datas) - { - var readOnlyMemory = (ReadOnlyMemory) data; - model.BasicPublish(exchange, routingKey, basicProperties, readOnlyMemory); - } - - model.WaitForConfirmsOrDie(); - } - - public void UnSubscribe(string queueName, string exchange, string routingKey, bool deleteQueue = false) - { - logger.LogInformation( - $"UnSubscribe queue: {queueName} exchange: {exchange} routingKey: {routingKey} deleteQueue: {deleteQueue}"); - using var channelAccessor = channelPool.Acquire(queueName); - if (deleteQueue) - { - var passive = channelAccessor.Channel.QueueDeclarePassive(queueName); - logger.LogInformation($"Queue {queueName} exists, deleting... {passive.MessageCount} messages in queue"); - channelAccessor.Channel.QueueDelete(queueName); - } - - channelPool.Release(queueName); - } - - public Task StartAsync(CancellationToken cancellationToken) - { - if (options.Value.Declarations.Count > 0) - { - using var channelAccessor = channelPool.Acquire(); - var channel = channelAccessor.Channel; - var decl = new DefaultDeclaration(channel); - options.Value.Declarations.ForEach(q => q.Invoke(decl)); - } - - foreach (var opt in options.Value.Consumes) - { - Subscribe(opt); - } - - return Task.CompletedTask; - } - - public Task StopAsync(CancellationToken cancellationToken) - { - channelPool.Dispose(); - return Task.CompletedTask; - } -} diff --git a/src/TrackCenter.Application/Rabbit/RabbitInstrumentation.cs b/src/TrackCenter.Application/Rabbit/RabbitInstrumentation.cs deleted file mode 100644 index 2afdc420..00000000 --- a/src/TrackCenter.Application/Rabbit/RabbitInstrumentation.cs +++ /dev/null @@ -1,21 +0,0 @@ -using System.Diagnostics; -using System.Diagnostics.Metrics; - -namespace TrackCenter.Application.Rabbit; - -public class RabbitInstrumentation -{ - public const string MeterName = "Fake.Rabbit.Meter"; - public const string ActivitySourceName = "Fake.Rabbit.ActivitySource"; - - static RabbitInstrumentation() - { - string version = typeof (RabbitInstrumentation).Assembly.GetName().Version?.ToString(); - Meter = new Meter(MeterName, version); - ActivitySource = new ActivitySource(ActivitySourceName, version); - } - - public static Meter Meter { get; } - - public static ActivitySource ActivitySource { get; } -} diff --git a/src/TrackCenter.Application/Rabbit/RabbitOptions.cs b/src/TrackCenter.Application/Rabbit/RabbitOptions.cs deleted file mode 100644 index 258f5ea7..00000000 --- a/src/TrackCenter.Application/Rabbit/RabbitOptions.cs +++ /dev/null @@ -1,84 +0,0 @@ -using RabbitMQ.Client; - -namespace TrackCenter.Application.Rabbit; - -public class RabbitOptions -{ - public List Consumes { get; set; } = new(); - - public List> Declarations { get; set; } = new(); -} - -public class ConsumeOptions -{ - /// - /// 队列名 - /// - public string Queue { get; set; } - - /// - /// 并行度 - /// - public ushort FetchCount { get; set; } = 1; - - /// - /// handle发生异常，是否重回队列 - /// - public bool FailedRequeue { get; set; } - - public Action Declaration { get; set; } -} - -public interface IDeclaration -{ - void QueueDeclare(string queue, bool durable, bool autoDelete = false, bool exclusive = false, - IDictionary arguments = null); - - void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete = false, - IDictionary arguments = null); - - void QueueBind(string queue, string exchange, string routingKey, - IDictionary arguments = null); -} - -public class DefaultDeclaration : IDeclaration -{ - private readonly IModel _chnl; - - // ReSharper disable once ConvertToPrimaryConstructor - public DefaultDeclaration(IModel chnl) - { - _chnl = chnl; - } - - public void QueueDeclare(string queue, bool durable, bool autoDelete = false, bool exclusive = false, - IDictionary arguments = null) - { - _chnl.QueueDeclare( - queue: queue, - durable: durable, - autoDelete: autoDelete, - exclusive: exclusive, - arguments: arguments); - } - - public void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete = false, - IDictionary arguments = null) - { - _chnl.ExchangeDeclare( - exchange: exchange, - type: type, - durable: durable, - autoDelete: autoDelete, - arguments: arguments); - } - - public void QueueBind(string queue, string exchange, string routingKey, IDictionary arguments) - { - _chnl.QueueBind( - queue: queue, - exchange: exchange, - routingKey: routingKey, - arguments: arguments); - } -} diff --git a/src/TrackCenter.Application/Services/Broadcasting/BroadcastingService.cs b/src/TrackCenter.Application/Services/Broadcasting/BroadcastingService.cs index 16468571..36c15a6d 100644 --- a/src/TrackCenter.Application/Services/Broadcasting/BroadcastingService.cs +++ b/src/TrackCenter.Application/Services/Broadcasting/BroadcastingService.cs @@ -1,74 +1,102 @@ using System.Text; +using Fake.RabbitMQ; using InterfaceForward.Domain.Shared.Helpers; +using RabbitMQ.Client; using TrackCenter.Application.Dtos.TrackPush; using TrackCenter.Application.Events; -using TrackCenter.Application.Rabbit; namespace TrackCenter.Application.Services.Broadcasting; /// -/// 广播服务 +/// 对外 Topic 广播（下游仍按 Newtonsoft PascalCase 契约消费，不经 Messaging 信封）。 /// -public class BroadcastingService : ITransientDependency +public class BroadcastingService(IRabbitMqChannelPool channelPool) : ITransientDependency { - private readonly RabbitClient _rabbitClient; - - public BroadcastingService(RabbitClient rabbitClient) + public void TrackChangePublicTopic(TrackChangeEto eto, string AppCode) { - _rabbitClient = rabbitClient; - } - - public void TrackChangePublicTopic(TrackChangeEto eto, string AppCode) - { - _rabbitClient.PublishByConfirmSelect( + PublishByConfirmSelect( GlobalConst.TrackChangeExchange, + ExchangeType.Topic, $"TrackChange.{AppCode}", Encoding.UTF8.GetBytes(eto.ToJson()), - prop => { prop.Persistent = true; } - ); + prop => { prop.Persistent = true; }); } /// /// 轨迹推送完成广播（批量） /// - /// - /// public void TrackPushCompletedPublicTopic(List list) { var data = list .Select(TrackPushRecordDtoToEvent) .Select(x => Encoding.UTF8.GetBytes(x.ToJson())); var log = list.First(); - _rabbitClient.PublishByConfirmSelect( + PublishByConfirmSelect( GlobalConst.TrackPushExchange, + ExchangeType.Topic, $"TrackPush.{log.AppKey}", data, - prop => { prop.Persistent = true; } - ); + prop => { prop.Persistent = true; }); } /// /// 轨迹推送完成广播 /// - /// - /// public void TrackPushCompletedPublicTopic(TrackPushRecordDto log) { var data = TrackPushRecordDtoToEvent(log); - _rabbitClient.PublishByConfirmSelect(GlobalConst.TrackPushExchange, $"TrackPush.{log.AppKey}", - Encoding.UTF8.GetBytes(data.ToJson()), prop => { prop.Persistent = true; }); + PublishByConfirmSelect( + GlobalConst.TrackPushExchange, + ExchangeType.Topic, + $"TrackPush.{log.AppKey}", + Encoding.UTF8.GetBytes(data.ToJson()), + prop => { prop.Persistent = true; }); } /// /// 轨迹推送失败广播 /// - /// public void TrackPushFailedPubilcTopic(TrackPushRecordDto log) { //暂时使用推送完成模型，后续可能会出现有区分度的模型 TrackPushCompletedPublicTopic(log); } + private void PublishByConfirmSelect( + string exchange, + string exchangeType, + string routingKey, + byte[] data, + Action? options = null) + { + using var channelAccessor = channelPool.Acquire(routingKey); + var channel = channelAccessor.Channel; + channel.ExchangeDeclare(exchange, exchangeType, durable: true); + var props = channel.CreateBasicProperties(); + options?.Invoke(props); + channel.ConfirmSelect(); + channel.BasicPublish(exchange, routingKey, props, data); + channel.WaitForConfirmsOrDie(); + } + + private void PublishByConfirmSelect( + string exchange, + string exchangeType, + string routingKey, + IEnumerable datas, + Action? options = null) + { + using var channelAccessor = channelPool.Acquire(routingKey); + var channel = channelAccessor.Channel; + channel.ExchangeDeclare(exchange, exchangeType, durable: true); + var props = channel.CreateBasicProperties(); + options?.Invoke(props); + channel.ConfirmSelect(); + foreach (var data in datas) + channel.BasicPublish(exchange, routingKey, props, data); + channel.WaitForConfirmsOrDie(); + } + private static TrackPushCompletedEvent TrackPushRecordDtoToEvent(TrackPushRecordDto log) { return new TrackPushCompletedEvent diff --git a/src/TrackCenter.Application/Services/ProvideServices/OrderService.Link.cs b/src/TrackCenter.Application/Services/ProvideServices/OrderService.Link.cs index 3371500e..89387578 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/OrderService.Link.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/OrderService.Link.cs @@ -1,5 +1,5 @@ using System.Data; -using Fake.EventBus.Distributed; +using Fake.Messaging; using Fake.UnitOfWork; using TrackCenter.Application.Dtos.ProvideServices; using TrackCenter.Application.Events; @@ -13,7 +13,7 @@ namespace TrackCenter.Application.Services.ProvideServices; /// public partial class OrderService { - private readonly IDistributedEventBus _eventBus = eventBus; + private readonly IMessagePublisher _messagePublisher = messagePublisher; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository = trackDetailsEsRepository; /// @@ -130,7 +130,7 @@ public partial class OrderService IsAdd = true // 存在则不新增 }; - await _eventBus.PublishAsync(fillTrackEvent); + await _messagePublisher.PublishAsync(fillTrackEvent); } } diff --git a/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs b/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs index 9211d0e2..6ccfa095 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/OrderService.cs @@ -1,4 +1,4 @@ -using Fake.EventBus.Distributed; +using Fake.Messaging; using TrackCenter.Application.Dtos.ProvideServices; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; @@ -17,7 +17,7 @@ public partial class OrderService( IOrderRepository orderRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, ITrackPullConfigRepository trackPullConfigRepository, - IDistributedEventBus eventBus, + IMessagePublisher messagePublisher, ITrackDetailsEsRepository trackDetailsEsRepository, IServiceProvider serviceProvider) : ApplicationServiceBase diff --git a/src/TrackCenter.Application/Services/ProvideServices/TrackPushService.cs b/src/TrackCenter.Application/Services/ProvideServices/TrackPushService.cs index f371e4ba..fe595f5d 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/TrackPushService.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/TrackPushService.cs @@ -1,5 +1,5 @@ using System.Data; -using Fake.EventBus.Distributed; +using Fake.Messaging; using Fake.UnitOfWork; using TrackCenter.Application.Dtos.ProvideServices; using TrackCenter.Application.Dtos.TrackOrder; @@ -20,20 +20,20 @@ namespace TrackCenter.Application.Services.ProvideServices; [Route("/ProviderService")] public class TrackPushService : ApplicationServiceBase { - private readonly IDistributedEventBus _eventBus; + private readonly IMessagePublisher _messagePublisher; private readonly IOrderRepository _orderRepository; private readonly ITrackPushConfigRepository _trackPushConfigRepository; private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; public TrackPushService(IOrderRepository orderRepository - , IDistributedEventBus eventBus + , IMessagePublisher messagePublisher , ITrackPushRecordEsRepository trackPushRecordEsRepository , ITrackPushConfigRepository trackPushConfigRepository, ITrackDetailsEsRepository trackDetailsEsRepository) { _orderRepository = orderRepository; - _eventBus = eventBus; + _messagePublisher = messagePublisher; _trackPushRecordEsRepository = trackPushRecordEsRepository; _trackPushConfigRepository = trackPushConfigRepository; _trackDetailsEsRepository = trackDetailsEsRepository; @@ -155,7 +155,7 @@ public class TrackPushService : ApplicationServiceBase } }; - await _eventBus.PublishAsync(trackPushEvent); + await _messagePublisher.PublishAsync(trackPushEvent); pushedOrderNos.Add(order.OrderNo); } diff --git a/src/TrackCenter.Application/Services/ProvideServices/TrackService.cs b/src/TrackCenter.Application/Services/ProvideServices/TrackService.cs index 0a4d22e0..fba94972 100644 --- a/src/TrackCenter.Application/Services/ProvideServices/TrackService.cs +++ b/src/TrackCenter.Application/Services/ProvideServices/TrackService.cs @@ -1,5 +1,5 @@ using System.Data; -using Fake.EventBus.Distributed; +using Fake.Messaging; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Dtos.ProvideServices; @@ -19,7 +19,7 @@ namespace TrackCenter.Application.Services.ProvideServices; [Route("/ProviderService")] public partial class TrackService( IOrderRepository orderRepository, - IDistributedEventBus eventBus, + IMessagePublisher messagePublisher, ITrackDetailsEsRepository trackDetailsEsRepository, ITrackContentRepository trackContentRepository) : ApplicationServiceBase @@ -90,7 +90,7 @@ public partial class TrackService( } // 删除轨迹(异步) - await eventBus.PublishAsync(new TrackDeleteEvent + await messagePublisher.PublishAsync(new TrackDeleteEvent { EsIds = esIds, OrderIds = new[] { orderId }, @@ -165,7 +165,7 @@ public partial class TrackService( AppKey = input.AppKey, OperationMan = CurrentUser.UserName }; - await eventBus.PublishAsync(@event); + await messagePublisher.PublishAsync(@event); } /// @@ -219,7 +219,7 @@ public partial class TrackService( } }; @event.OperationMan = CurrentUser.UserName; - await eventBus.PublishAsync(@event); + await messagePublisher.PublishAsync(@event); } /// @@ -292,7 +292,7 @@ public partial class TrackService( ForcedPush = input.ForcedPush, IsToMq = input.IsToMq, }; - await eventBus.PublishAsync(@event); + await messagePublisher.PublishAsync(@event); } } diff --git a/src/TrackCenter.Application/Services/Warning/ReturnWarningService.cs b/src/TrackCenter.Application/Services/Warning/ReturnWarningService.cs index 368d0c6d..c420c49f 100644 --- a/src/TrackCenter.Application/Services/Warning/ReturnWarningService.cs +++ b/src/TrackCenter.Application/Services/Warning/ReturnWarningService.cs @@ -1,5 +1,5 @@ using System.Data; -using Fake.EventBus.Distributed; +using Fake.Messaging; using Fake.Domain.Dtos; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; @@ -27,7 +27,7 @@ public class ReturnWarningService : ApplicationServiceBase private readonly IReturnWarningHandler _returnWarningHandler; private readonly IWarningRuleRepository _warningRuleRepository; private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; - private readonly IDistributedEventBus _eventBus; + private readonly IMessagePublisher _messagePublisher; private readonly TrackStatusNameDictCache _trackStatusNameDictCache; private readonly AppNameDictCache _appNameDictCache; @@ -36,7 +36,7 @@ public class ReturnWarningService : ApplicationServiceBase IReturnWarningHandler returnWarningHandler, IWarningRuleRepository warningRuleRepository, IWarningDetailsEsRepository warningDetailsEsRepository, - IDistributedEventBus eventBus, + IMessagePublisher messagePublisher, TrackStatusNameDictCache trackStatusNameDictCache, AppNameDictCache appNameDictCache ) @@ -45,7 +45,7 @@ public class ReturnWarningService : ApplicationServiceBase _returnWarningHandler = returnWarningHandler; _warningRuleRepository = warningRuleRepository; _warningDetailsEsRepository = warningDetailsEsRepository; - _eventBus = eventBus; + _messagePublisher = messagePublisher; _trackStatusNameDictCache = trackStatusNameDictCache; _appNameDictCache = appNameDictCache; } @@ -279,7 +279,7 @@ public class ReturnWarningService : ApplicationServiceBase } }; - await _eventBus.PublishAsync(trackPushEvent); + await _messagePublisher.PublishAsync(trackPushEvent); } return $"即将进行{filtered.Count}个回传推送！"; diff --git a/src/TrackCenter.Application/TrackCenter.Application.csproj b/src/TrackCenter.Application/TrackCenter.Application.csproj index ccab7abc..43dd17cb 100644 --- a/src/TrackCenter.Application/TrackCenter.Application.csproj +++ b/src/TrackCenter.Application/TrackCenter.Application.csproj @@ -9,8 +9,7 @@ - - + diff --git a/src/TrackCenter.Application/TrackCenterApplicationModule.cs b/src/TrackCenter.Application/TrackCenterApplicationModule.cs index 4c0de9ef..dd7535d1 100644 --- a/src/TrackCenter.Application/TrackCenterApplicationModule.cs +++ b/src/TrackCenter.Application/TrackCenterApplicationModule.cs @@ -1,28 +1,26 @@ using System.Diagnostics; using System.Net; using Fake; -using Fake.EventBus.RabbitMQ; +using Fake.Messaging; +using Fake.Messaging.RabbitMQ; using Fake.Modularity; using Fake.ObjectMapping.AutoMapper; using Fake.ObjectStorage; -using Fake.RabbitMQ; using FreeRedis; using Hangfire; using Hangfire.Redis.StackExchange; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; -using RabbitMQ.Client; -using TrackCenter.Application.Kafka; using TrackCenter.Application.HostServices; +using TrackCenter.Application.Kafka; +using TrackCenter.Application.Messaging; using TrackCenter.Application.Options; -using TrackCenter.Application.Rabbit; namespace TrackCenter.Application; -[DependsOn(typeof(FakeRabbitMqModule))] [DependsOn(typeof(FakeObjectMappingAutoMapperModule))] -[DependsOn(typeof(FakeEventBusRabbitMqModule))] +[DependsOn(typeof(FakeMessagingRabbitMqModule))] [DependsOn(typeof(FakeObjectStorageTencentCosModule))] [DependsOn(typeof(TrackCenterDomainModule))] public class TrackCenterApplicationModule : FakeModule @@ -36,50 +34,27 @@ public class TrackCenterApplicationModule : FakeModule options.ScanProfiles(); }); + // Messaging：独立队列；PrefetchCount=0 → ProcessorCount；接口通可按 TrackApp:TrackRefreshFetchCount 覆盖 + context.Services.Configure(configuration.GetSection(MessagingOptions.SectionName)); + context.Services.Configure(o => + { + configuration.GetSection(RabbitMessageOptions.SectionName).Bind(o); + + var refreshFetch = configuration.GetValue("TrackApp:TrackRefreshFetchCount"); + if (refreshFetch > 0) + { + o.Handlers[nameof(TrackRefreshEventHandler)] = + new RabbitMessageHandlerOptions { PrefetchCount = refreshFetch }; + o.Handlers[nameof(TrackPushCompletedEventHandler)] = + new RabbitMessageHandlerOptions { PrefetchCount = refreshFetch }; + } + }); + var redisOptions = configuration.GetSection("Redis").Get(); if (redisOptions == null) throw new FakeException("未找到Redis相关配置，请检查配置文件中是否包含名为Redis的节点"); RedisHelper.Init(redisOptions, showLogInfo: Debugger.IsAttached); - context.Services.AddHostedService(); - context.Services.Configure(options => - { - options.Declarations = new List>() - { - o => o.ExchangeDeclare(GlobalConst.TrackChangeExchange, ExchangeType.Topic, true), - o => o.ExchangeDeclare(GlobalConst.TrackPushExchange, ExchangeType.Topic, true) - }; - options.Consumes = new List - { - new() - { - Queue = GlobalConst.TrackRefreshSubscribe, - FetchCount = configuration.GetValue("TrackApp:TrackRefreshFetchCount"), - FailedRequeue = false, - Declaration = o => - { - o.ExchangeDeclare(GlobalConst.InterfaceRelayExchange, ExchangeType.Direct, true); - o.QueueDeclare(GlobalConst.TrackRefreshSubscribe, true); - o.QueueBind(GlobalConst.TrackRefreshSubscribe, GlobalConst.InterfaceRelayExchange, - GlobalConst.TrackRefreshSubscribe); - } - }, - new() - { - Queue = GlobalConst.TrackPushCompletedSubscribe, - FetchCount = configuration.GetValue("TrackApp:TrackRefreshFetchCount"), - FailedRequeue = false, - Declaration = o => - { - o.ExchangeDeclare(GlobalConst.InterfaceRelayExchange, ExchangeType.Direct, true); - o.QueueDeclare(GlobalConst.TrackPushCompletedSubscribe, true); - o.QueueBind(GlobalConst.TrackPushCompletedSubscribe, GlobalConst.InterfaceRelayExchange, - GlobalConst.TrackPushCompletedSubscribe); - } - } - }; - }); - context.Services.Configure(configuration.GetSection("TrackApp")); context.Services.Configure( configuration.GetSection("InterfaceRelayParamsConfig"));