msumshk/TakeoutSaaS.TenantApi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: 升级依赖并优化种子

Browse Source
This commit is contained in:
MSuMshk
2025-12-04 17:30:09 +08:00
parent f50d2d4bb2
commit 1d7836a173
16 changed files with 163 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/Modules/TakeoutSaaS.Module.Messaging/Services/RabbitMqConnectionFactory.cs
View File

@@ -12,7 +12,7 @@ public sealed class RabbitMqConnectionFactory(IOptionsMonitor<RabbitMqOptions> o
/// <summary>
/// 创建连接。
/// </summary>
public IConnection CreateConnection()
public Task<IConnection> CreateConnectionAsync(CancellationToken cancellationToken = default)
{
var options = optionsMonitor.CurrentValue;
var factory = new ConnectionFactory
@@ -21,10 +21,9 @@ public sealed class RabbitMqConnectionFactory(IOptionsMonitor<RabbitMqOptions> o
Port = options.Port,
UserName = options.Username,
Password = options.Password,
VirtualHost = options.VirtualHost,
DispatchConsumersAsync = true
VirtualHost = options.VirtualHost
};
return factory.CreateConnection();
return factory.CreateConnectionAsync(cancellationToken);
}
}

27
src/Modules/TakeoutSaaS.Module.Messaging/Services/RabbitMqMessagePublisher.cs
View File

@@ -14,41 +14,42 @@ public sealed class RabbitMqMessagePublisher(RabbitMqConnectionFactory connectio
: IMessagePublisher, IAsyncDisposable
{
private IConnection? _connection;
private IModel? _channel;
private IChannel? _channel;
private bool _disposed;
/// <inheritdoc />
public Task PublishAsync<T>(string routingKey, T message, CancellationToken cancellationToken = default)
public async Task PublishAsync<T>(string routingKey, T message, CancellationToken cancellationToken = default)
{
// 1. 确保通道可用
EnsureChannel();
await EnsureChannelAsync(cancellationToken);
var options = optionsMonitor.CurrentValue;
var channel = _channel ?? throw new InvalidOperationException("RabbitMQ channel is not available.");
// 2. 声明交换机
channel.ExchangeDeclare(options.Exchange, options.ExchangeType, durable: true, autoDelete: false);
await channel.ExchangeDeclareAsync(options.Exchange, options.ExchangeType, durable: true, autoDelete: false, cancellationToken: cancellationToken);
// 3. 序列化消息并设置属性
var body = serializer.Serialize(message);
var props = channel.CreateBasicProperties();
props.ContentType = "application/json";
props.DeliveryMode = 2;
props.MessageId = Guid.NewGuid().ToString("N");
var props = new BasicProperties
{
ContentType = "application/json",
DeliveryMode = DeliveryModes.Persistent,
MessageId = Guid.NewGuid().ToString("N")
};
// 4. 发布消息
channel.BasicPublish(options.Exchange, routingKey, props, body);
await channel.BasicPublishAsync(options.Exchange, routingKey, mandatory: false, props, body, cancellationToken);
logger.LogDebug("发布消息到交换机 {Exchange} RoutingKey {RoutingKey}", options.Exchange, routingKey);
return Task.CompletedTask;
}
private void EnsureChannel()
private async Task EnsureChannelAsync(CancellationToken cancellationToken)
{
if (_channel != null && _channel.IsOpen)
{
return;
}
_connection ??= connectionFactory.CreateConnection();
_channel = _connection.CreateModel();
_connection ??= await connectionFactory.CreateConnectionAsync(cancellationToken);
_channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
}
/// <summary>

29
src/Modules/TakeoutSaaS.Module.Messaging/Services/RabbitMqMessageSubscriber.cs
View File

@@ -15,32 +15,32 @@ public sealed class RabbitMqMessageSubscriber(RabbitMqConnectionFactory connecti
: IMessageSubscriber
{
private IConnection? _connection;
private IModel? _channel;
private IChannel? _channel;
private bool _disposed;
/// <inheritdoc />
public async Task SubscribeAsync<T>(string queue, string routingKey, Func<T, CancellationToken, Task<bool>> handler, CancellationToken cancellationToken = default)
{
// 1. 确保通道可用
EnsureChannel();
await EnsureChannelAsync(cancellationToken);
var options = optionsMonitor.CurrentValue;
var channel = _channel ?? throw new InvalidOperationException("RabbitMQ channel is not available.");
// 2. 声明交换机、队列及绑定
channel.ExchangeDeclare(options.Exchange, options.ExchangeType, durable: true, autoDelete: false);
channel.QueueDeclare(queue, durable: true, exclusive: false, autoDelete: false);
channel.QueueBind(queue, options.Exchange, routingKey);
channel.BasicQos(0, options.PrefetchCount, global: false);
await channel.ExchangeDeclareAsync(options.Exchange, options.ExchangeType, durable: true, autoDelete: false, cancellationToken: cancellationToken);
await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false, cancellationToken: cancellationToken);
await channel.QueueBindAsync(queue, options.Exchange, routingKey, cancellationToken: cancellationToken);
await channel.BasicQosAsync(0, options.PrefetchCount, global: false, cancellationToken: cancellationToken);
// 3. 设置消费者回调
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.Received += async (_, ea) =>
consumer.ReceivedAsync += async (_, ea) =>
{
var message = serializer.Deserialize<T>(ea.Body.ToArray());
if (message == null)
{
channel.BasicAck(ea.DeliveryTag, multiple: false);
await channel.BasicAckAsync(ea.DeliveryTag, multiple: false, cancellationToken);
return;
}
@@ -56,28 +56,27 @@ public sealed class RabbitMqMessageSubscriber(RabbitMqConnectionFactory connecti
if (success)
{
channel.BasicAck(ea.DeliveryTag, multiple: false);
await channel.BasicAckAsync(ea.DeliveryTag, multiple: false, cancellationToken);
}
else
{
channel.BasicNack(ea.DeliveryTag, multiple: false, requeue: false);
await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: false, cancellationToken);
}
};
// 4. 开始消费
channel.BasicConsume(queue, autoAck: false, consumer);
await Task.CompletedTask.ConfigureAwait(false);
await channel.BasicConsumeAsync(queue, autoAck: false, consumer, cancellationToken);
}
private void EnsureChannel()
private async Task EnsureChannelAsync(CancellationToken cancellationToken)
{
if (_channel != null && _channel.IsOpen)
{
return;
}
_connection ??= connectionFactory.CreateConnection();
_channel = _connection.CreateModel();
_connection ??= await connectionFactory.CreateConnectionAsync(cancellationToken);
_channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
}
/// <inheritdoc />

14
src/Modules/TakeoutSaaS.Module.Messaging/TakeoutSaaS.Module.Messaging.csproj
View File

@@ -5,13 +5,13 @@
<ImplicitUsings>enable</ImplicitUsings>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="8.0.0" />
<PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" Version="8.0.0" />
<PackageReference Include="RabbitMQ.Client" Version="6.6.0" />
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Options" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" Version="10.0.0" />
<PackageReference Include="RabbitMQ.Client" Version="7.2.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Core\TakeoutSaaS.Shared.Abstractions\TakeoutSaaS.Shared.Abstractions.csproj" />