From 039cbba3d1feb11b8e69767872661499f603982f Mon Sep 17 00:00:00 2001 From: xiaolipro <2357729423@qq.com> Date: Fri, 17 Jul 2026 17:43:13 +0800 Subject: [PATCH] =?UTF-8?q?feat=EF=BC=9Afake?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../DataPerm/EffectiveScopeBuilder.cs | 30 ++-- src/Oms.Application/Helpers/RedisHelper.cs | 83 +++++++++++ .../Messaging/OrderBatchJobPublisher.cs | 80 +++-------- .../Messaging/OrderBatchQueueInitializer.cs | 24 +--- .../Messaging/OrderBatchQueueInspector.cs | 80 ++++------- .../Messaging/OrderBatchRabbitMqOptions.cs | 13 +- src/Oms.Application/Oms.Application.csproj | 4 +- src/Oms.Application/OmsApplicationModule.cs | 38 ++++- .../Services/Plugin/PluginBrowserService.cs | 15 +- .../OrderBatch/OrderBatchCreateConsumer.cs | 132 ++---------------- .../OrderBatchIntegrationConsumer.cs | 132 ++---------------- .../OrderBatch/OrderBatchQueueConsumerBase.cs | 71 +++++----- src/Oms.E2E.Tests/Support/OrderE2eTestHost.cs | 2 + .../Support/SimpleAdminApiTestHost.cs | 5 +- src/SimpleAdmin.Api/SimpleAdmin.Api.csproj | 1 - .../Services/WeChatLoginSessionStore.cs | 14 +- .../WeChat/WeChatServiceExtensions.cs | 10 -- .../appsettings.Development.json | 26 +++- 18 files changed, 273 insertions(+), 487 deletions(-) create mode 100644 src/Oms.Application/Helpers/RedisHelper.cs diff --git a/src/Oms.Application/DataPerm/EffectiveScopeBuilder.cs b/src/Oms.Application/DataPerm/EffectiveScopeBuilder.cs index d0da508..ac9efaa 100644 --- a/src/Oms.Application/DataPerm/EffectiveScopeBuilder.cs +++ b/src/Oms.Application/DataPerm/EffectiveScopeBuilder.cs @@ -1,32 +1,20 @@ using Fake.Application; using Fake.DependencyInjection; -using Microsoft.Extensions.Caching.Distributed; -using System.Text.Json; +using Oms.Application.Helpers; namespace Oms.Application.DataPerm; -public class DataPermContextCache(IDistributedCache cache) : IScopedDependency +public class DataPermContextCache : IScopedDependency { - private static readonly JsonSerializerOptions JsonOptions = new() - { - PropertyNameCaseInsensitive = true - }; - private static readonly TimeSpan CacheTtl = TimeSpan.FromMinutes(30); public async Task GetAsync(Guid userId, string dataResourceCode, CancellationToken cancellationToken = default) { - var bytes = await cache.GetAsync(BuildKey(userId, dataResourceCode), cancellationToken); - if (bytes == null || bytes.Length == 0) - { - return null; - } - try { - return JsonSerializer.Deserialize(bytes, JsonOptions); + return await RedisHelper.Client.GetAsync(BuildKey(userId, dataResourceCode)); } - catch (JsonException) + catch { // 旧缓存或类型变更导致反序列化失败时重建 return null; @@ -35,19 +23,17 @@ public class DataPermContextCache(IDistributedCache cache) : IScopedDependency public Task SetAsync(DataScopeContext context, CancellationToken cancellationToken = default) { - var bytes = JsonSerializer.SerializeToUtf8Bytes(context, JsonOptions); - return cache.SetAsync( + return RedisHelper.Client.SetAsync( BuildKey(context.UserId, context.DataResourceCode), - bytes, - new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = CacheTtl }, - cancellationToken); + context, + (int)CacheTtl.TotalSeconds); } public async Task InvalidateUserAsync(Guid userId, CancellationToken cancellationToken = default) { foreach (var resourceCode in DataPermConstants.FilterableFieldsByResource.Keys) { - await cache.RemoveAsync(BuildKey(userId, resourceCode), cancellationToken); + await RedisHelper.Client.DelAsync(BuildKey(userId, resourceCode)); } } diff --git a/src/Oms.Application/Helpers/RedisHelper.cs b/src/Oms.Application/Helpers/RedisHelper.cs new file mode 100644 index 0000000..a52ef4b --- /dev/null +++ b/src/Oms.Application/Helpers/RedisHelper.cs @@ -0,0 +1,83 @@ +using FreeRedis; +using Newtonsoft.Json; +using Serilog; + +namespace Oms.Application.Helpers; + +public static class RedisHelper +{ + private static readonly object Lock = new(); + private static RedisClient? _client; + private static ConnectionStringBuilder Options = null!; + + #region 初始化方法 + + internal static void Init(ConnectionStringBuilder options, bool showLogInfo) + { + Options = options; + + if (_client == null) + { + lock (Lock) + { + if (_client == null) + { + _client = new RedisClient(options); + _client.Serialize = obj => JsonConvert.SerializeObject(obj); + _client.Deserialize = (json, type) => JsonConvert.DeserializeObject(json, type); + if (showLogInfo) + _client.Notice += (s, e) => Log.Information(Environment.NewLine + e.Log); + } + } + } + } + + /// + /// 创建一个新的客户端 + /// + public static RedisClient Create(Action? action, EventHandler? notice) + { + var builder = new ConnectionStringBuilder + { + Database = Options.Database, + ExitAutoDisposePool = Options.ExitAutoDisposePool, + ClientName = Options.ClientName, + ConnectTimeout = Options.ConnectTimeout, + Encoding = Options.Encoding, + Host = Options.Host, + IdleTimeout = Options.IdleTimeout, + MaxPoolSize = Options.MaxPoolSize, + MinPoolSize = Options.MinPoolSize, + Password = Options.Password, + Prefix = Options.Prefix, + Protocol = Options.Protocol, + ReceiveTimeout = Options.ReceiveTimeout, + Retry = Options.Retry, + SendTimeout = Options.SendTimeout, + Ssl = Options.Ssl, + SubscribeReadbytes = Options.SubscribeReadbytes, + User = Options.User + }; + + action?.Invoke(builder); + + var client = new RedisClient(builder); + + client.Serialize = obj => JsonConvert.SerializeObject(obj); + client.Deserialize = (json, type) => JsonConvert.DeserializeObject(json, type); + if (notice != null) + { + client.Notice += notice; + } + + return client; + } + + #endregion + + #region 返回客户端类 + + public static RedisClient Client => _client!; + + #endregion +} diff --git a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs b/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs index f6b453f..8e7c6db 100644 --- a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs +++ b/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs @@ -1,23 +1,23 @@ using System.Text; +using Fake.RabbitMQ; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Newtonsoft.Json; -using RabbitMQ.Client; namespace Oms.Application.Messaging; -public sealed class OrderBatchJobPublisher : IOrderBatchJobPublisher, IDisposable +public sealed class OrderBatchJobPublisher : IOrderBatchJobPublisher { + private readonly IRabbitMqChannelPool _channelPool; private readonly OrderBatchRabbitMqOptions _options; private readonly ILogger _logger; - private readonly object _sync = new(); - private IConnection? _connection; - private bool _disposed; public OrderBatchJobPublisher( + IRabbitMqChannelPool channelPool, IOptions options, ILogger logger) { + _channelPool = channelPool; _options = options.Value; _logger = logger; } @@ -34,28 +34,25 @@ public sealed class OrderBatchJobPublisher : IOrderBatchJobPublisher, IDisposabl var routingKey = OrderBatchQueueTopology.ResolveRoutingKey(message.Action); var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message)); - lock (_sync) + using var channelAccessor = _channelPool.Acquire("oms-order-batch-publisher"); + var channel = channelAccessor.Channel; + var props = channel.CreateBasicProperties(); + props.Persistent = true; + props.ContentType = "application/json"; + props.MessageId = message.MessageId; + props.Headers = new Dictionary { - EnsureConnection(); - using var channel = _connection!.CreateModel(); - var props = channel.CreateBasicProperties(); - props.Persistent = true; - props.ContentType = "application/json"; - props.MessageId = message.MessageId; - props.Headers = new Dictionary - { - ["action"] = message.Action, - ["jobId"] = message.JobId, - ["jobItemId"] = message.JobItemId, - }; + ["action"] = message.Action, + ["jobId"] = message.JobId, + ["jobItemId"] = message.JobItemId, + }; - channel.BasicPublish( - exchange: OrderBatchQueueTopology.Exchange, - routingKey: routingKey, - mandatory: false, - basicProperties: props, - body: body); - } + channel.BasicPublish( + exchange: OrderBatchQueueTopology.Exchange, + routingKey: routingKey, + mandatory: false, + basicProperties: props, + body: body); _logger.LogDebug( "Published order batch message {MessageId} action={Action} routingKey={RoutingKey}", @@ -65,37 +62,4 @@ public sealed class OrderBatchJobPublisher : IOrderBatchJobPublisher, IDisposabl return Task.CompletedTask; } - - private void EnsureConnection() - { - if (_connection is { IsOpen: true }) - { - return; - } - - var factory = new ConnectionFactory - { - HostName = _options.HostName, - Port = _options.Port, - UserName = _options.UserName, - Password = _options.Password, - VirtualHost = _options.VirtualHost, - DispatchConsumersAsync = true, - AutomaticRecoveryEnabled = true, - }; - - _connection?.Dispose(); - _connection = factory.CreateConnection("oms-order-batch-publisher"); - } - - public void Dispose() - { - if (_disposed) - { - return; - } - - _disposed = true; - _connection?.Dispose(); - } } diff --git a/src/Oms.Application/Messaging/OrderBatchQueueInitializer.cs b/src/Oms.Application/Messaging/OrderBatchQueueInitializer.cs index 06ce284..387cb38 100644 --- a/src/Oms.Application/Messaging/OrderBatchQueueInitializer.cs +++ b/src/Oms.Application/Messaging/OrderBatchQueueInitializer.cs @@ -1,3 +1,4 @@ +using Fake.RabbitMQ; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -10,13 +11,16 @@ namespace Oms.Application.Messaging; /// public sealed class OrderBatchQueueInitializer : IHostedService { + private readonly IRabbitMqChannelPool _channelPool; private readonly OrderBatchRabbitMqOptions _options; private readonly ILogger _logger; public OrderBatchQueueInitializer( + IRabbitMqChannelPool channelPool, IOptions options, ILogger logger) { + _channelPool = channelPool; _options = options.Value; _logger = logger; } @@ -31,8 +35,8 @@ public sealed class OrderBatchQueueInitializer : IHostedService try { - using var connection = CreateConnection(); - using var channel = connection.CreateModel(); + using var channelAccessor = _channelPool.Acquire("oms-order-batch-init"); + var channel = channelAccessor.Channel; channel.ExchangeDeclare(OrderBatchQueueTopology.DeadLetterExchange, ExchangeType.Topic, durable: true); channel.QueueDeclare( @@ -71,20 +75,4 @@ public sealed class OrderBatchQueueInitializer : IHostedService } public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; - - private IConnection CreateConnection() - { - var factory = new ConnectionFactory - { - HostName = _options.HostName, - Port = _options.Port, - UserName = _options.UserName, - Password = _options.Password, - VirtualHost = _options.VirtualHost, - DispatchConsumersAsync = true, - AutomaticRecoveryEnabled = true, - }; - - return factory.CreateConnection("oms-order-batch-init"); - } } diff --git a/src/Oms.Application/Messaging/OrderBatchQueueInspector.cs b/src/Oms.Application/Messaging/OrderBatchQueueInspector.cs index 2540f8f..ac5dbb0 100644 --- a/src/Oms.Application/Messaging/OrderBatchQueueInspector.cs +++ b/src/Oms.Application/Messaging/OrderBatchQueueInspector.cs @@ -1,3 +1,4 @@ +using Fake.RabbitMQ; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Oms.Application.Dtos.Orders; @@ -12,17 +13,18 @@ public interface IOrderBatchQueueInspector Task GetStatsAsync(CancellationToken cancellationToken = default); } -public sealed class OrderBatchQueueInspector : IOrderBatchQueueInspector, IDisposable +public sealed class OrderBatchQueueInspector : IOrderBatchQueueInspector { + private readonly IRabbitMqChannelPool _channelPool; private readonly OrderBatchRabbitMqOptions _options; private readonly ILogger _logger; - private readonly object _sync = new(); - private IConnection? _connection; public OrderBatchQueueInspector( + IRabbitMqChannelPool channelPool, IOptions options, ILogger logger) { + _channelPool = channelPool; _options = options.Value; _logger = logger; } @@ -40,31 +42,29 @@ public sealed class OrderBatchQueueInspector : IOrderBatchQueueInspector, IDispo }); } - lock (_sync) + using var channelAccessor = _channelPool.Acquire("oms-order-batch-inspector"); + var channel = channelAccessor.Channel; + var queues = new List { - EnsureConnection(); - using var channel = _connection!.CreateModel(); - var queues = new List - { - InspectQueue(channel, OrderBatchQueueTopology.QueueDeadLetter), - InspectQueue(channel, OrderBatchQueueTopology.QueueIntegration), - InspectQueue(channel, OrderBatchQueueTopology.QueueCreate), - InspectQueue(channel, OrderBatchQueueTopology.QueueAccount), - InspectQueue(channel, OrderBatchQueueTopology.QueueState), - InspectQueue(channel, OrderBatchQueueTopology.QueueException), - InspectQueue(channel, OrderBatchQueueTopology.QueueDocument), - }; + InspectQueue(channel, OrderBatchQueueTopology.QueueDeadLetter), + InspectQueue(channel, OrderBatchQueueTopology.QueueIntegration), + InspectQueue(channel, OrderBatchQueueTopology.QueueCreate), + InspectQueue(channel, OrderBatchQueueTopology.QueueAccount), + InspectQueue(channel, OrderBatchQueueTopology.QueueState), + InspectQueue(channel, OrderBatchQueueTopology.QueueException), + InspectQueue(channel, OrderBatchQueueTopology.QueueDocument), + }; - var dlxCount = queues.First(x => x.QueueName == OrderBatchQueueTopology.QueueDeadLetter).MessageCount; - return Task.FromResult(new OrderBatchDlxStatsOutput - { - Enabled = true, - DeadLetterMessageCount = dlxCount, - DlxAlertThreshold = _options.DlxAlertThreshold, - Alert = dlxCount >= _options.DlxAlertThreshold, - Queues = queues, - }); - } + var dlxCount = queues.First(x => x.QueueName == OrderBatchQueueTopology.QueueDeadLetter).MessageCount; + _logger.LogDebug("Order batch queue stats inspected, dlxMessages={DeadLetterCount}", dlxCount); + return Task.FromResult(new OrderBatchDlxStatsOutput + { + Enabled = true, + DeadLetterMessageCount = dlxCount, + DlxAlertThreshold = _options.DlxAlertThreshold, + Alert = dlxCount >= _options.DlxAlertThreshold, + Queues = queues, + }); } private static OrderBatchQueueStatOutput InspectQueue(IModel channel, string queueName) @@ -84,32 +84,4 @@ public sealed class OrderBatchQueueInspector : IOrderBatchQueueInspector, IDispo return new OrderBatchQueueStatOutput { QueueName = queueName }; } } - - private void EnsureConnection() - { - if (_connection is { IsOpen: true }) - { - return; - } - - var factory = new ConnectionFactory - { - HostName = _options.HostName, - Port = _options.Port, - UserName = _options.UserName, - Password = _options.Password, - VirtualHost = _options.VirtualHost, - DispatchConsumersAsync = true, - AutomaticRecoveryEnabled = true, - }; - - _connection?.Dispose(); - _connection = factory.CreateConnection("oms-order-batch-inspector"); - _logger.LogDebug("Order batch queue inspector connected."); - } - - public void Dispose() - { - _connection?.Dispose(); - } } diff --git a/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs b/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs index 9e47486..e2b821f 100644 --- a/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs +++ b/src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs @@ -1,21 +1,14 @@ namespace Oms.Application.Messaging; +/// +/// 订单批量队列业务配置。连接信息使用 Fake 标准 RabbitMQ 配置节（FakeRabbitMqOptions）。 +/// public class OrderBatchRabbitMqOptions { public const string SectionName = "OrderBatch:RabbitMq"; public bool Enabled { get; set; } - public string HostName { get; set; } = "localhost"; - - public int Port { get; set; } = 5672; - - public string UserName { get; set; } = "oms"; - - public string Password { get; set; } = "oms123"; - - public string VirtualHost { get; set; } = "/oms"; - public ushort PrefetchCount { get; set; } = 5; public int MaxRetryCount { get; set; } = 3; diff --git a/src/Oms.Application/Oms.Application.csproj b/src/Oms.Application/Oms.Application.csproj index efc0437..51fa64c 100644 --- a/src/Oms.Application/Oms.Application.csproj +++ b/src/Oms.Application/Oms.Application.csproj @@ -19,12 +19,14 @@ + + - + diff --git a/src/Oms.Application/OmsApplicationModule.cs b/src/Oms.Application/OmsApplicationModule.cs index 305e199..b2b65cc 100644 --- a/src/Oms.Application/OmsApplicationModule.cs +++ b/src/Oms.Application/OmsApplicationModule.cs @@ -1,7 +1,12 @@ +using System.Diagnostics; +using Fake; using Fake.Modularity; using Fake.ObjectMapping.AutoMapper; -using Fake.Security.Claims; +using Fake.RabbitMQ; +using FreeRedis; +using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Oms.Application.Helpers; using Oms.Application.Integrations.SmallBagTrack; using Oms.Application.Integrations.Itfx; using Oms.Application.Integrations.PlatformShop; @@ -16,23 +21,42 @@ namespace Oms.Application; /// [DependsOn( typeof(OmsDomainModule), - typeof(FakeObjectMappingAutoMapperModule) + typeof(FakeObjectMappingAutoMapperModule), + typeof(FakeRabbitMqModule) )] public class OmsApplicationModule : FakeModule { /// public override void ConfigureServices(ServiceConfigurationContext context) { + var configuration = context.Services.GetConfiguration(); + context.Services.Configure(options => { options.ScanProfiles(); }); - context.Services.AddOmsSmallBagTrack(context.Services.GetConfiguration()); - context.Services.AddOmsItfxIntegration(context.Services.GetConfiguration()); - context.Services.AddOmsPlatformShopIntegration(context.Services.GetConfiguration()); - context.Services.AddOmsOrderBatchMessaging(context.Services.GetConfiguration()); + context.Services.PostConfigure(options => + { + foreach (var factory in options.Connections.Values) + { + factory.DispatchConsumersAsync = true; + factory.AutomaticRecoveryEnabled = true; + } + }); + + var redisOptions = configuration.GetSection("Redis").Get(); + if (redisOptions == null || string.IsNullOrWhiteSpace(redisOptions.Host)) + { + throw new FakeException("未找到Redis相关配置，请检查配置文件中是否包含名为Redis的节点（且包含 Host）"); + } + + RedisHelper.Init(redisOptions, showLogInfo: Debugger.IsAttached); + + context.Services.AddOmsSmallBagTrack(configuration); + context.Services.AddOmsItfxIntegration(configuration); + context.Services.AddOmsPlatformShopIntegration(configuration); + context.Services.AddOmsOrderBatchMessaging(configuration); context.Services.AddXhsLlmClient(); - context.Services.AddDistributedMemoryCache(); context.Services.AddHostedService(); } } diff --git a/src/Oms.Application/Services/Plugin/PluginBrowserService.cs b/src/Oms.Application/Services/Plugin/PluginBrowserService.cs index 1aefa6b..bf15f6a 100644 --- a/src/Oms.Application/Services/Plugin/PluginBrowserService.cs +++ b/src/Oms.Application/Services/Plugin/PluginBrowserService.cs @@ -1,7 +1,6 @@ -using System.Text; using Fake.Application; using Microsoft.AspNetCore.Mvc; -using Microsoft.Extensions.Caching.Distributed; +using Oms.Application.Helpers; using Oms.Application.Shared; using Oms.Domain.CustomersAggregate.Repositories; using Oms.Domain.SystemAggregate.Repositories; @@ -22,7 +21,6 @@ public class PluginBrowserService : ApplicationService private ICParcelRepository ParcelRepository => LazyServiceProvider.GetRequiredService(); private IWarehouseAddressRepository WarehouseAddressRepository => LazyServiceProvider.GetRequiredService(); - private IDistributedCache Cache => LazyServiceProvider.GetRequiredService(); /// /// 缓存采集草稿，返回 key（确认页 getPackage 使用）。 @@ -40,10 +38,7 @@ public class PluginBrowserService : ApplicationService } var key = $"{resolvedCustomerId}{DateTimeOffset.UtcNow.ToUnixTimeSeconds()}{Random.Shared.Next(100, 999)}"; - await Cache.SetAsync( - CacheKey(key), - Encoding.UTF8.GetBytes(payload!), - new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = DraftTtl }); + await RedisHelper.Client.SetAsync(CacheKey(key), payload!, (int)DraftTtl.TotalSeconds); return key; } @@ -59,13 +54,13 @@ public class PluginBrowserService : ApplicationService throw new BusinessException("key 不能为空"); } - var bytes = await Cache.GetAsync(CacheKey(key.Trim())); - if (bytes == null || bytes.Length == 0) + var payload = await RedisHelper.Client.GetAsync(CacheKey(key.Trim())); + if (string.IsNullOrEmpty(payload)) { throw new BusinessException("包裹草稿不存在或已过期"); } - return Encoding.UTF8.GetString(bytes); + return payload; } /// diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs b/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs index 88a4d78..8a52b02 100644 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs +++ b/src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs @@ -1,136 +1,22 @@ +using Fake.RabbitMQ; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Oms.Application.Messaging; -using Oms.Application.Services.Orders.Batch; -using RabbitMQ.Client; -using RabbitMQ.Client.Events; namespace Oms.Application.Workers.OrderBatch; /// /// 消费创建类队列（Excel 导入行、小包批量创建）。 /// -public sealed class OrderBatchCreateConsumer : BackgroundService +public sealed class OrderBatchCreateConsumer( + IRabbitMqChannelPool channelPool, + IOptions options, + IServiceScopeFactory scopeFactory, + ILogger logger) + : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) { - private readonly OrderBatchRabbitMqOptions _options; - private readonly IServiceScopeFactory _scopeFactory; - private readonly ILogger _logger; - private IConnection? _connection; - private IModel? _channel; + protected override string QueueName => OrderBatchQueueTopology.QueueCreate; - public OrderBatchCreateConsumer( - IOptions options, - IServiceScopeFactory scopeFactory, - ILogger logger) - { - _options = options.Value; - _scopeFactory = scopeFactory; - _logger = logger; - } - - protected override Task ExecuteAsync(CancellationToken stoppingToken) - { - if (!_options.Enabled) - { - _logger.LogInformation("Order batch RabbitMQ disabled, create consumer not started."); - return Task.CompletedTask; - } - - var factory = new ConnectionFactory - { - HostName = _options.HostName, - Port = _options.Port, - UserName = _options.UserName, - Password = _options.Password, - VirtualHost = _options.VirtualHost, - DispatchConsumersAsync = true, - AutomaticRecoveryEnabled = true, - }; - - _connection = factory.CreateConnection("oms-order-batch-create-consumer"); - _channel = _connection.CreateModel(); - _channel.BasicQos(0, _options.PrefetchCount, false); - - var consumer = new AsyncEventingBasicConsumer(_channel); - consumer.Received += OnReceivedAsync; - - _channel.BasicConsume( - queue: OrderBatchQueueTopology.QueueCreate, - autoAck: false, - consumer: consumer); - - _logger.LogInformation( - "Order batch create consumer started on {Queue}.", - OrderBatchQueueTopology.QueueCreate); - - stoppingToken.Register(() => - { - try - { - _channel?.Close(); - _connection?.Close(); - } - catch (Exception ex) - { - _logger.LogDebug(ex, "Error closing RabbitMQ create consumer."); - } - }); - - return Task.CompletedTask; - } - - private async Task OnReceivedAsync(object sender, BasicDeliverEventArgs ea) - { - if (_channel == null) - { - return; - } - - try - { - var message = OrderBatchJobProcessor.Deserialize(ea.Body); - if (message == null) - { - _channel.BasicAck(ea.DeliveryTag, false); - return; - } - - using var scope = _scopeFactory.CreateScope(); - var processor = scope.ServiceProvider.GetRequiredService(); - var result = await processor.ProcessAsync(message); - - switch (result.Kind) - { - case OrderBatchJobProcessor.ProcessResultKind.Ack: - _channel.BasicAck(ea.DeliveryTag, false); - break; - case OrderBatchJobProcessor.ProcessResultKind.Reject: - if (result.Requeue && message.RetryCount < _options.MaxRetryCount) - { - message.RetryCount++; - _channel.BasicNack(ea.DeliveryTag, false, true); - } - else - { - _channel.BasicAck(ea.DeliveryTag, false); - } - - break; - } - } - catch (Exception ex) - { - _logger.LogError(ex, "Order batch create consumer failed."); - _channel.BasicNack(ea.DeliveryTag, false, true); - } - } - - public override void Dispose() - { - _channel?.Dispose(); - _connection?.Dispose(); - base.Dispose(); - } + protected override bool UseRetryLimit => true; } diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs b/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs index 2cf5803..ddaa095 100644 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs +++ b/src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs @@ -1,136 +1,22 @@ +using Fake.RabbitMQ; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Oms.Application.Messaging; -using Oms.Application.Services.Orders.Batch; -using RabbitMQ.Client; -using RabbitMQ.Client.Events; namespace Oms.Application.Workers.OrderBatch; /// /// 消费外部集成队列（轨迹同步、下游预报）。 /// -public sealed class OrderBatchIntegrationConsumer : BackgroundService +public sealed class OrderBatchIntegrationConsumer( + IRabbitMqChannelPool channelPool, + IOptions options, + IServiceScopeFactory scopeFactory, + ILogger logger) + : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) { - private readonly OrderBatchRabbitMqOptions _options; - private readonly IServiceScopeFactory _scopeFactory; - private readonly ILogger _logger; - private IConnection? _connection; - private IModel? _channel; + protected override string QueueName => OrderBatchQueueTopology.QueueIntegration; - public OrderBatchIntegrationConsumer( - IOptions options, - IServiceScopeFactory scopeFactory, - ILogger logger) - { - _options = options.Value; - _scopeFactory = scopeFactory; - _logger = logger; - } - - protected override Task ExecuteAsync(CancellationToken stoppingToken) - { - if (!_options.Enabled) - { - _logger.LogInformation("Order batch RabbitMQ disabled, integration consumer not started."); - return Task.CompletedTask; - } - - var factory = new ConnectionFactory - { - HostName = _options.HostName, - Port = _options.Port, - UserName = _options.UserName, - Password = _options.Password, - VirtualHost = _options.VirtualHost, - DispatchConsumersAsync = true, - AutomaticRecoveryEnabled = true, - }; - - _connection = factory.CreateConnection("oms-order-batch-integration-consumer"); - _channel = _connection.CreateModel(); - _channel.BasicQos(0, _options.PrefetchCount, false); - - var consumer = new AsyncEventingBasicConsumer(_channel); - consumer.Received += OnReceivedAsync; - - _channel.BasicConsume( - queue: OrderBatchQueueTopology.QueueIntegration, - autoAck: false, - consumer: consumer); - - _logger.LogInformation( - "Order batch integration consumer started on {Queue}.", - OrderBatchQueueTopology.QueueIntegration); - - stoppingToken.Register(() => - { - try - { - _channel?.Close(); - _connection?.Close(); - } - catch (Exception ex) - { - _logger.LogDebug(ex, "Error closing RabbitMQ consumer."); - } - }); - - return Task.CompletedTask; - } - - private async Task OnReceivedAsync(object sender, BasicDeliverEventArgs ea) - { - if (_channel == null) - { - return; - } - - try - { - var message = OrderBatchJobProcessor.Deserialize(ea.Body); - if (message == null) - { - _channel.BasicAck(ea.DeliveryTag, false); - return; - } - - using var scope = _scopeFactory.CreateScope(); - var processor = scope.ServiceProvider.GetRequiredService(); - var result = await processor.ProcessAsync(message); - - switch (result.Kind) - { - case OrderBatchJobProcessor.ProcessResultKind.Ack: - _channel.BasicAck(ea.DeliveryTag, false); - break; - case OrderBatchJobProcessor.ProcessResultKind.Reject: - if (result.Requeue && message.RetryCount < _options.MaxRetryCount) - { - message.RetryCount++; - _channel.BasicNack(ea.DeliveryTag, false, true); - } - else - { - _channel.BasicAck(ea.DeliveryTag, false); - } - - break; - } - } - catch (Exception ex) - { - _logger.LogError(ex, "Order batch integration consumer failed."); - _channel.BasicNack(ea.DeliveryTag, false, true); - } - } - - public override void Dispose() - { - _channel?.Dispose(); - _connection?.Dispose(); - base.Dispose(); - } + protected override bool UseRetryLimit => true; } diff --git a/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs b/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs index afa52c0..6845f2d 100644 --- a/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs +++ b/src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs @@ -1,3 +1,4 @@ +using Fake.RabbitMQ; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; @@ -11,17 +12,20 @@ namespace Oms.Application.Workers.OrderBatch; public abstract class OrderBatchQueueConsumerBase : BackgroundService { + private readonly IRabbitMqChannelPool _channelPool; private readonly OrderBatchRabbitMqOptions _options; private readonly IServiceScopeFactory _scopeFactory; private readonly ILogger _logger; - private IConnection? _connection; + private IChannelAccessor? _channelAccessor; private IModel? _channel; protected OrderBatchQueueConsumerBase( + IRabbitMqChannelPool channelPool, IOptions options, IServiceScopeFactory scopeFactory, ILogger logger) { + _channelPool = channelPool; _options = options.Value; _scopeFactory = scopeFactory; _logger = logger; @@ -29,7 +33,10 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService protected abstract string QueueName { get; } - protected abstract string ConnectionName { get; } + /// + /// 为 true 时，Reject+Requeue 受 限制。 + /// + protected virtual bool UseRetryLimit => false; protected override Task ExecuteAsync(CancellationToken stoppingToken) { @@ -41,19 +48,8 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService return Task.CompletedTask; } - var factory = new ConnectionFactory - { - HostName = _options.HostName, - Port = _options.Port, - UserName = _options.UserName, - Password = _options.Password, - VirtualHost = _options.VirtualHost, - DispatchConsumersAsync = true, - AutomaticRecoveryEnabled = true, - }; - - _connection = factory.CreateConnection(ConnectionName); - _channel = _connection.CreateModel(); + _channelAccessor = _channelPool.Acquire(QueueName); + _channel = _channelAccessor.Channel; _channel.BasicQos(0, _options.PrefetchCount, false); var consumer = new AsyncEventingBasicConsumer(_channel); @@ -67,12 +63,11 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService { try { - _channel?.Close(); - _connection?.Close(); + _channelPool.Release(QueueName); } catch (Exception ex) { - _logger.LogDebug(ex, "Error closing RabbitMQ consumer {Queue}.", QueueName); + _logger.LogDebug(ex, "Error releasing RabbitMQ consumer {Queue}.", QueueName); } }); @@ -107,7 +102,22 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService case OrderBatchJobProcessor.ProcessResultKind.Reject: if (result.Requeue) { - _channel.BasicNack(ea.DeliveryTag, false, true); + if (UseRetryLimit) + { + if (message.RetryCount < _options.MaxRetryCount) + { + message.RetryCount++; + _channel.BasicNack(ea.DeliveryTag, false, true); + } + else + { + _channel.BasicAck(ea.DeliveryTag, false); + } + } + else + { + _channel.BasicNack(ea.DeliveryTag, false, true); + } } else { @@ -126,52 +136,47 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService public override void Dispose() { - _channel?.Dispose(); - _connection?.Dispose(); + _channelAccessor?.Dispose(); base.Dispose(); } } public sealed class OrderBatchAccountConsumer( + IRabbitMqChannelPool channelPool, IOptions options, IServiceScopeFactory scopeFactory, ILogger logger) - : OrderBatchQueueConsumerBase(options, scopeFactory, logger) + : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) { protected override string QueueName => OrderBatchQueueTopology.QueueAccount; - - protected override string ConnectionName => "oms-order-batch-account-consumer"; } public sealed class OrderBatchStateConsumer( + IRabbitMqChannelPool channelPool, IOptions options, IServiceScopeFactory scopeFactory, ILogger logger) - : OrderBatchQueueConsumerBase(options, scopeFactory, logger) + : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) { protected override string QueueName => OrderBatchQueueTopology.QueueState; - - protected override string ConnectionName => "oms-order-batch-state-consumer"; } public sealed class OrderBatchExceptionConsumer( + IRabbitMqChannelPool channelPool, IOptions options, IServiceScopeFactory scopeFactory, ILogger logger) - : OrderBatchQueueConsumerBase(options, scopeFactory, logger) + : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) { protected override string QueueName => OrderBatchQueueTopology.QueueException; - - protected override string ConnectionName => "oms-order-batch-exception-consumer"; } public sealed class OrderBatchDocumentConsumer( + IRabbitMqChannelPool channelPool, IOptions options, IServiceScopeFactory scopeFactory, ILogger logger) - : OrderBatchQueueConsumerBase(options, scopeFactory, logger) + : OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger) { protected override string QueueName => OrderBatchQueueTopology.QueueDocument; - - protected override string ConnectionName => "oms-order-batch-document-consumer"; } diff --git a/src/Oms.E2E.Tests/Support/OrderE2eTestHost.cs b/src/Oms.E2E.Tests/Support/OrderE2eTestHost.cs index 8504510..c980fda 100644 --- a/src/Oms.E2E.Tests/Support/OrderE2eTestHost.cs +++ b/src/Oms.E2E.Tests/Support/OrderE2eTestHost.cs @@ -49,6 +49,8 @@ internal sealed class OrderE2eTestHost : IAsyncDisposable ["FeiShu:EnableFeiShuExceptionSubscribe"] = "false", ["FeiShuNotice:Webhook"] = "https://example.com/webhook", ["FeiShuNotice:EnableFeiShuExceptionSubscribe"] = "false", + ["Redis:Host"] = "127.0.0.1:6379", + ["Redis:Database"] = "0", ["OrderBatch:RabbitMq:Enabled"] = "false", ["Oms:TrackService:Enabled"] = "false", ["Oms:TrackService:BaseUrl"] = "http://127.0.0.1:9090", diff --git a/src/SimpleAdmin.Api.IntegrationTests/Support/SimpleAdminApiTestHost.cs b/src/SimpleAdmin.Api.IntegrationTests/Support/SimpleAdminApiTestHost.cs index 49bf56d..3002612 100644 --- a/src/SimpleAdmin.Api.IntegrationTests/Support/SimpleAdminApiTestHost.cs +++ b/src/SimpleAdmin.Api.IntegrationTests/Support/SimpleAdminApiTestHost.cs @@ -73,7 +73,10 @@ internal sealed class SimpleAdminApiTestHost : IAsyncDisposable ["FeiShu:Webhook"] = "https://example.com/webhook", ["FeiShu:EnableFeiShuExceptionSubscribe"] = "false", ["FeiShuNotice:Webhook"] = "https://example.com/webhook", - ["FeiShuNotice:EnableFeiShuExceptionSubscribe"] = "false" + ["FeiShuNotice:EnableFeiShuExceptionSubscribe"] = "false", + ["Redis:Host"] = "127.0.0.1:6379", + ["Redis:Database"] = "0", + ["OrderBatch:RabbitMq:Enabled"] = "false" }); }, environmentName: "Testing"); diff --git a/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj b/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj index 18db10c..6de202a 100644 --- a/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj +++ b/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj @@ -14,7 +14,6 @@ - diff --git a/src/SimpleAdmin.Api/WeChat/Services/WeChatLoginSessionStore.cs b/src/SimpleAdmin.Api/WeChat/Services/WeChatLoginSessionStore.cs index 81e8a04..e53c964 100644 --- a/src/SimpleAdmin.Api/WeChat/Services/WeChatLoginSessionStore.cs +++ b/src/SimpleAdmin.Api/WeChat/Services/WeChatLoginSessionStore.cs @@ -1,9 +1,9 @@ using System.Text.Json; -using Microsoft.Extensions.Caching.Distributed; +using Oms.Application.Helpers; namespace SimpleAdmin.Api.WeChat.Services; -public class WeChatLoginSessionStore(IDistributedCache cache) : IWeChatLoginSessionStore +public class WeChatLoginSessionStore : IWeChatLoginSessionStore { private const string KeyPrefix = "oms:wechat:login:"; @@ -18,19 +18,15 @@ public class WeChatLoginSessionStore(IDistributedCache cache) : IWeChatLoginSess TimeSpan ttl, CancellationToken cancellationToken = default) { - var options = new DistributedCacheEntryOptions - { - AbsoluteExpirationRelativeToNow = ttl - }; var json = JsonSerializer.Serialize(payload, JsonOptions); - return cache.SetStringAsync(BuildKey(loginKey), json, options, cancellationToken); + return RedisHelper.Client.SetAsync(BuildKey(loginKey), json, (int)ttl.TotalSeconds); } public async Task GetSessionAsync( string loginKey, CancellationToken cancellationToken = default) { - var raw = await cache.GetStringAsync(BuildKey(loginKey), cancellationToken); + var raw = await RedisHelper.Client.GetAsync(BuildKey(loginKey)); if (string.IsNullOrWhiteSpace(raw)) { return null; @@ -47,7 +43,7 @@ public class WeChatLoginSessionStore(IDistributedCache cache) : IWeChatLoginSess public Task RemoveAsync(string loginKey, CancellationToken cancellationToken = default) { - return cache.RemoveAsync(BuildKey(loginKey), cancellationToken); + return RedisHelper.Client.DelAsync(BuildKey(loginKey)); } private static string BuildKey(string loginKey) => $"{KeyPrefix}{loginKey}"; diff --git a/src/SimpleAdmin.Api/WeChat/WeChatServiceExtensions.cs b/src/SimpleAdmin.Api/WeChat/WeChatServiceExtensions.cs index 0009fdd..4a730d1 100644 --- a/src/SimpleAdmin.Api/WeChat/WeChatServiceExtensions.cs +++ b/src/SimpleAdmin.Api/WeChat/WeChatServiceExtensions.cs @@ -13,16 +13,6 @@ public static class WeChatServiceExtensions services.AddHttpClient(nameof(WeChatMpApiClient)); services.AddMemoryCache(); - var redisConnection = configuration.GetConnectionString("Redis") ?? configuration["Redis:ConnectionString"]; - if (!string.IsNullOrWhiteSpace(redisConnection)) - { - services.AddStackExchangeRedisCache(options => options.Configuration = redisConnection); - } - else - { - services.AddDistributedMemoryCache(); - } - services.AddScoped(); services.AddSingleton(); services.AddTransient(); diff --git a/src/SimpleAdmin.Api/appsettings.Development.json b/src/SimpleAdmin.Api/appsettings.Development.json index a873cb3..4edc5b8 100644 --- a/src/SimpleAdmin.Api/appsettings.Development.json +++ b/src/SimpleAdmin.Api/appsettings.Development.json @@ -1,5 +1,5 @@ { - "UseLocalConfigs": true, + "UseLocalConfigs": false, "Logging": { "LogLevel": { "Default": "Information", @@ -40,7 +40,13 @@ "EnableFeiShuExceptionSubscribe": false }, "Redis": { - "ConnectionString": "" + "Host": "127.0.0.1:6379", + "Password": "", + "Database": 0, + "MaxPoolSize": 50, + "MinPoolSize": 1, + "IdleTimeout": 20000, + "ConnectTimeout": 5000 }, "WeChat": { "Mp": { @@ -90,14 +96,20 @@ } } }, + "RabbitMQ": { + "Connections": { + "Default": { + "HostName": "127.0.0.1", + "Port": 5672, + "UserName": "oms", + "Password": "oms123", + "VirtualHost": "/oms" + } + } + }, "OrderBatch": { "RabbitMq": { "Enabled": false, - "HostName": "127.0.0.1", - "Port": 5672, - "UserName": "oms", - "Password": "oms123", - "VirtualHost": "/oms", "PrefetchCount": 5, "MaxRetryCount": 3 }