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

feat：fake

Browse Source
This commit is contained in:
xiaolipro 2026-07-17 17:43:13 +08:00
parent 54bba99153
commit 039cbba3d1
18 changed files with 273 additions and 487 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/Oms.Application/DataPerm/EffectiveScopeBuilder.cs
View File

@ -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<DataScopeContext?> 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<DataScopeContext>(bytes, JsonOptions);
return await RedisHelper.Client.GetAsync<DataScopeContext>(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));
}
}

83
src/Oms.Application/Helpers/RedisHelper.cs Normal file
View File

@ -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);
}
}
}
}
/// <summary>
/// 创建一个新的客户端
/// </summary>
public static RedisClient Create(Action<ConnectionStringBuilder>? action, EventHandler<NoticeEventArgs>? 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
}

50
src/Oms.Application/Messaging/OrderBatchJobPublisher.cs
View File

@ -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<OrderBatchJobPublisher> _logger;
private readonly object _sync = new();
private IConnection? _connection;
private bool _disposed;
public OrderBatchJobPublisher(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchJobPublisher> logger)
{
_channelPool = channelPool;
_options = options.Value;
_logger = logger;
}
@ -34,10 +34,8 @@ public sealed class OrderBatchJobPublisher : IOrderBatchJobPublisher, IDisposabl
var routingKey = OrderBatchQueueTopology.ResolveRoutingKey(message.Action);
var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message));
lock (_sync)
{
EnsureConnection();
using var channel = _connection!.CreateModel();
using var channelAccessor = _channelPool.Acquire("oms-order-batch-publisher");
var channel = channelAccessor.Channel;
var props = channel.CreateBasicProperties();
props.Persistent = true;
props.ContentType = "application/json";
@ -55,7 +53,6 @@ public sealed class OrderBatchJobPublisher : IOrderBatchJobPublisher, IDisposabl
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();
}
}

24
src/Oms.Application/Messaging/OrderBatchQueueInitializer.cs
View File

@ -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;
/// </summary>
public sealed class OrderBatchQueueInitializer : IHostedService
{
private readonly IRabbitMqChannelPool _channelPool;
private readonly OrderBatchRabbitMqOptions _options;
private readonly ILogger<OrderBatchQueueInitializer> _logger;
public OrderBatchQueueInitializer(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchQueueInitializer> 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");
}
}

44
src/Oms.Application/Messaging/OrderBatchQueueInspector.cs
View File

@ -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<OrderBatchDlxStatsOutput> 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<OrderBatchQueueInspector> _logger;
private readonly object _sync = new();
private IConnection? _connection;
public OrderBatchQueueInspector(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
ILogger<OrderBatchQueueInspector> logger)
{
_channelPool = channelPool;
_options = options.Value;
_logger = logger;
}
@ -40,10 +42,8 @@ public sealed class OrderBatchQueueInspector : IOrderBatchQueueInspector, IDispo
});
}
lock (_sync)
{
EnsureConnection();
using var channel = _connection!.CreateModel();
using var channelAccessor = _channelPool.Acquire("oms-order-batch-inspector");
var channel = channelAccessor.Channel;
var queues = new List<OrderBatchQueueStatOutput>
{
InspectQueue(channel, OrderBatchQueueTopology.QueueDeadLetter),
@ -56,6 +56,7 @@ public sealed class OrderBatchQueueInspector : IOrderBatchQueueInspector, IDispo
};
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,
@ -65,7 +66,6 @@ public sealed class OrderBatchQueueInspector : IOrderBatchQueueInspector, IDispo
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();
}
}

13
src/Oms.Application/Messaging/OrderBatchRabbitMqOptions.cs
View File

@ -1,21 +1,14 @@
namespace Oms.Application.Messaging;
/// <summary>
/// 订单批量队列业务配置。连接信息使用 Fake 标准 <c>RabbitMQ</c> 配置节（FakeRabbitMqOptions）。
/// </summary>
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;

4
src/Oms.Application/Oms.Application.csproj
View File

@ -19,12 +19,14 @@
<ItemGroup>
<PackageReference Include="Fake.ObjectMapping.AutoMapper" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.RabbitMQ" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Rbac.Application" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Rbac.Infrastructure" Version="$(FakePackageVersion)" />
<PackageReference Include="DocumentFormat.OpenXml" Version="3.0.2" />
<PackageReference Include="FreeRedis" Version="1.5.5" />
<PackageReference Include="MailKit" Version="4.17.0" />
<PackageReference Include="MiniExcel" Version="1.43.1" />
<PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
<PackageReference Include="Serilog" Version="4.2.0" />
</ItemGroup>
<ItemGroup>

38
src/Oms.Application/OmsApplicationModule.cs
View File

@ -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;
/// </summary>
[DependsOn(
typeof(OmsDomainModule),
typeof(FakeObjectMappingAutoMapperModule)
typeof(FakeObjectMappingAutoMapperModule),
typeof(FakeRabbitMqModule)
)]
public class OmsApplicationModule : FakeModule
{
/// <inheritdoc />
public override void ConfigureServices(ServiceConfigurationContext context)
{
var configuration = context.Services.GetConfiguration();
context.Services.Configure<FakeAutoMapperOptions>(options =>
{
options.ScanProfiles<OmsApplicationModule>();
});
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<FakeRabbitMqOptions>(options =>
{
foreach (var factory in options.Connections.Values)
{
factory.DispatchConsumersAsync = true;
factory.AutomaticRecoveryEnabled = true;
}
});
var redisOptions = configuration.GetSection("Redis").Get<ConnectionStringBuilder>();
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<XhsScheduledPublishWorker>();
}
}

15
src/Oms.Application/Services/Plugin/PluginBrowserService.cs
View File

@ -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<ICParcelRepository>();
private IWarehouseAddressRepository WarehouseAddressRepository =>
LazyServiceProvider.GetRequiredService<IWarehouseAddressRepository>();
private IDistributedCache Cache => LazyServiceProvider.GetRequiredService<IDistributedCache>();
/// <summary>
/// 缓存采集草稿，返回 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;
}
/// <summary>

128
src/Oms.Application/Workers/OrderBatch/OrderBatchCreateConsumer.cs
View File

@ -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;
/// <summary>
/// 消费创建类队列（Excel 导入行、小包批量创建）。
/// </summary>
public sealed class OrderBatchCreateConsumer : BackgroundService
{
private readonly OrderBatchRabbitMqOptions _options;
private readonly IServiceScopeFactory _scopeFactory;
private readonly ILogger<OrderBatchCreateConsumer> _logger;
private IConnection? _connection;
private IModel? _channel;
public OrderBatchCreateConsumer(
public sealed class OrderBatchCreateConsumer(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchCreateConsumer> logger)
{
_options = options.Value;
_scopeFactory = scopeFactory;
_logger = logger;
}
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueCreate;
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<OrderBatchJobProcessor>();
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;
}

128
src/Oms.Application/Workers/OrderBatch/OrderBatchIntegrationConsumer.cs
View File

@ -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;
/// <summary>
/// 消费外部集成队列（轨迹同步、下游预报）。
/// </summary>
public sealed class OrderBatchIntegrationConsumer : BackgroundService
{
private readonly OrderBatchRabbitMqOptions _options;
private readonly IServiceScopeFactory _scopeFactory;
private readonly ILogger<OrderBatchIntegrationConsumer> _logger;
private IConnection? _connection;
private IModel? _channel;
public OrderBatchIntegrationConsumer(
public sealed class OrderBatchIntegrationConsumer(
IRabbitMqChannelPool channelPool,
IOptions<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchIntegrationConsumer> logger)
{
_options = options.Value;
_scopeFactory = scopeFactory;
_logger = logger;
}
: OrderBatchQueueConsumerBase(channelPool, options, scopeFactory, logger)
{
protected override string QueueName => OrderBatchQueueTopology.QueueIntegration;
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<OrderBatchJobProcessor>();
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;
}

69
src/Oms.Application/Workers/OrderBatch/OrderBatchQueueConsumerBase.cs
View File

@ -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<OrderBatchRabbitMqOptions> 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; }
/// <summary>
/// 为 true 时，Reject+Requeue 受 <see cref="OrderBatchRabbitMqOptions.MaxRetryCount"/> 限制。
/// </summary>
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,12 +102,27 @@ public abstract class OrderBatchQueueConsumerBase : BackgroundService
case OrderBatchJobProcessor.ProcessResultKind.Reject:
if (result.Requeue)
{
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
{
_channel.BasicAck(ea.DeliveryTag, false);
}
break;
}
@ -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<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchAccountConsumer> 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<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchStateConsumer> 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<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchExceptionConsumer> 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<OrderBatchRabbitMqOptions> options,
IServiceScopeFactory scopeFactory,
ILogger<OrderBatchDocumentConsumer> 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";
}

2
src/Oms.E2E.Tests/Support/OrderE2eTestHost.cs
View File

@ -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",

5
src/SimpleAdmin.Api.IntegrationTests/Support/SimpleAdminApiTestHost.cs
View File

@ -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");

1
src/SimpleAdmin.Api/SimpleAdmin.Api.csproj
View File

@ -14,7 +14,6 @@
<PackageReference Include="Fake.Consul" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Rbac.Application" Version="$(FakePackageVersion)" />
<PackageReference Include="Fake.Serilog" Version="$(FakePackageVersion)" />
<PackageReference Include="Microsoft.Extensions.Caching.StackExchangeRedis" Version="8.0.11" />
<PackageReference Include="Serilog.AspNetCore" Version="8.0.3"/>
</ItemGroup>

14
src/SimpleAdmin.Api/WeChat/Services/WeChatLoginSessionStore.cs
View File

@ -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<WeChatLoginSessionPayload?> 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}";

10
src/SimpleAdmin.Api/WeChat/WeChatServiceExtensions.cs
View File

@ -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<IWeChatMpApiClient, WeChatMpApiClient>();
services.AddSingleton<IWeChatLoginSessionStore, WeChatLoginSessionStore>();
services.AddTransient<IWeChatEventHandler, WeChatEventHandler>();

24
src/SimpleAdmin.Api/appsettings.Development.json
View File

@ -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 @@
}
}
},
"OrderBatch": {
"RabbitMq": {
"Enabled": false,
"RabbitMQ": {
"Connections": {
"Default": {
"HostName": "127.0.0.1",
"Port": 5672,
"UserName": "oms",
"Password": "oms123",
"VirtualHost": "/oms",
"VirtualHost": "/oms"
}
}
},
"OrderBatch": {
"RabbitMq": {
"Enabled": false,
"PrefetchCount": 5,
"MaxRetryCount": 3
}