glz/QYZH.InteractiveMagazine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor,feat: 批量代码重构与新增业务模块

Browse Source 
1. 重命名枚举项与实体字段，修正类型引用
2. 新增IsNull扩展方法与多项字符串处理扩展
3. 新增大量业务DTO、服务接口与枚举定义
4. 重构RabbitMQ服务实现，替换旧版消息队列组件
5. 优化签到服务的宠物喂养事务逻辑
6. 移除冗余的项目引用与旧版消息队列代码
7. 新增Excel导出、导入模板相关工具方法
This commit is contained in:
glz
2026-06-11 17:01:24 +08:00
parent ae4cd627df
commit e493c85d08
57 changed files with 4477 additions and 399 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
QYZH.InteractiveMagazine.Infrastructure/Autofac/AutofacExtension.cs
View File

@ -2,7 +2,6 @@
using Autofac.Extensions.DependencyInjection;
using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.Configuration;
using QYZH.InteractiveMagazine.Infrastructure.MessageQueue;
using QYZH.InteractiveMagazine.Models.Settings;
using RabbitMQ.Client;
using StackExchange.Redis;
@ -25,31 +24,7 @@ namespace QYZH.InteractiveMagazine.Infrastructure.Autofacs
var source = friendlyName.Split('.');
var assemblyNames = string.Join(".", source.Take(source.Length - 1));
containerBuilder.RegisterModule(new AutofacModuleRegister(assemblyNames));
InitializeRabbitMQ(c.Configuration, containerBuilder);
});
}
private static void InitializeRabbitMQ(IConfiguration configuration, ContainerBuilder containerBuilder)
{
var rabbitMQSettings = configuration.GetSection("RabbitMQSettings").Get<RabbitMQSettings>();
if (rabbitMQSettings != null && !string.IsNullOrWhiteSpace(rabbitMQSettings.HostName))
{
var factory = new ConnectionFactory
{
HostName = rabbitMQSettings.HostName,
Port = rabbitMQSettings.Port,
UserName = rabbitMQSettings.UserName ?? string.Empty,
Password = rabbitMQSettings.Password ?? string.Empty,
VirtualHost = rabbitMQSettings.VirtualHost ?? string.Empty
};
var connection = factory.CreateConnectionAsync().GetAwaiter().GetResult();
containerBuilder.RegisterInstance(connection).As<IConnection>().SingleInstance();
containerBuilder.RegisterType<RabbitMQPublisher>().InstancePerDependency();
}
}
}
}

72
QYZH.InteractiveMagazine.Infrastructure/MessageQueue/RabbitMQConsumer.cs
View File

@ -1,72 +0,0 @@
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
namespace QYZH.InteractiveMagazine.Infrastructure.MessageQueue;
/// <summary>
/// RabbitMQ消息消费者基类
/// </summary>
public abstract class RabbitMQConsumer : IDisposable
{
private readonly IConnection _connection;
private readonly ILogger<RabbitMQConsumer> _logger;
private IChannel? _channel;
private AsyncEventingBasicConsumer? _consumer;
/// <summary>
/// 构造函数
/// </summary>
/// <param name="connection">RabbitMQ连接</param>
/// <param name="logger">日志记录器</param>
protected RabbitMQConsumer(IConnection connection, ILogger<RabbitMQConsumer> logger)
{
_connection = connection;
_logger = logger;
}
/// <summary>
/// 启动消费
/// </summary>
/// <param name="queueName">队列名称</param>
/// <param name="handleMessage">消息处理委托</param>
public async Task StartConsume(string queueName, Func<string, Task> handleMessage)
{
_channel = await _connection.CreateChannelAsync();
_consumer = new AsyncEventingBasicConsumer(_channel);
_consumer.ReceivedAsync += async (model, ea) =>
{
try
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
await handleMessage(message);
await _channel.BasicAckAsync(ea.DeliveryTag, false);
_logger.LogInformation("消息消费成功 | 队列: {QueueName} | 消息: {Message}", queueName, message);
}
catch (Exception ex)
{
_logger.LogError(ex, "消息消费失败 | 队列: {QueueName}", queueName);
await _channel.BasicNackAsync(ea.DeliveryTag, false, true);
}
};
await _channel.BasicConsumeAsync(queue: queueName, autoAck: false, consumer: _consumer);
_logger.LogInformation("开始消费消息 | 队列: {QueueName}", queueName);
}
/// <summary>
/// 释放资源
/// </summary>
public void Dispose()
{
_channel?.DisposeAsync().GetAwaiter().GetResult();
GC.SuppressFinalize(this);
}
}

47
QYZH.InteractiveMagazine.Infrastructure/MessageQueue/RabbitMQPublisher.cs
View File

@ -1,47 +0,0 @@
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using System.Text;
namespace QYZH.InteractiveMagazine.Infrastructure.MessageQueue;
/// <summary>
/// RabbitMQ消息发布器
/// </summary>
public class RabbitMQPublisher
{
private readonly IConnection _connection;
private readonly ILogger<RabbitMQPublisher> _logger;
/// <summary>
/// 构造函数
/// </summary>
/// <param name="connection">RabbitMQ连接</param>
/// <param name="logger">日志记录器</param>
public RabbitMQPublisher(IConnection connection, ILogger<RabbitMQPublisher> logger)
{
_connection = connection;
_logger = logger;
}
/// <summary>
/// 发布消息
/// </summary>
/// <param name="exchange">交换机名称</param>
/// <param name="routingKey">路由键</param>
/// <param name="message">消息内容</param>
public async Task PublishMessage(string exchange, string routingKey, string message)
{
await using var channel = await _connection.CreateChannelAsync();
var body = Encoding.UTF8.GetBytes(message);
await channel.BasicPublishAsync(
exchange: exchange,
routingKey: routingKey,
body: body
);
_logger.LogInformation("消息已发布 | 交换机: {Exchange} | 路由键: {RoutingKey} | 消息: {Message}",
exchange, routingKey, message);
}
}

4
QYZH.InteractiveMagazine.Infrastructure/OSS/OssService.cs
View File

@ -217,7 +217,7 @@ namespace QYZH.InteractiveMagazine.Infrastructure.OSS
//var del = _ossResourceRepository.Delete(w => keys.Contains(w.Path));
//AppException.ThrowIf(!del, "删除资源失败");
//BusinessException.ThrowIf(!del, "删除资源失败");
return result.Keys.Count() == keys.Count;
@ -286,7 +286,7 @@ namespace QYZH.InteractiveMagazine.Infrastructure.OSS
// Path = targetObject,
//});
//AppException.ThrowIf(data.IsNull(), "资源添加失败");
//BusinessException.ThrowIf(data.IsNull(), "资源添加失败");
var req = new CopyObjectRequest(_ossOption.BucketName, sourceObject, _ossOption.BucketName, targetObject)
{

14
QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/IRabbitMQService.cs Normal file
View File

@ -0,0 +1,14 @@
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
namespace QYZH.InteractiveMagazine.Infrastructure.RabbitMQ
{
public interface IRabbitMQService
{
Task<bool> SendAsync(RabbitMQSendParam param, CancellationToken cancellationToken = default);
Task<bool> SendBatchAsync(IEnumerable<RabbitMQSendParam> @params, CancellationToken cancellationToken = default);
Task ReceiveAsync(string queueName, Func<IChannel, BasicDeliverEventArgs, Task> callback, CancellationToken cancellationToken = default);
}
}

62
QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/RabbitMQConnection.cs Normal file
View File

@ -0,0 +1,62 @@
using RabbitMQ.Client;
namespace QYZH.InteractiveMagazine.Infrastructure.RabbitMQ
{
public interface IRabbitMQConnection : IDisposable
{
Task<IChannel> CreateChannel();
}
public class RabbitMQConnection : IRabbitMQConnection
{
private readonly ConnectionFactory _factory;
private readonly IConnection _connection;
private bool _isDisposed;
public RabbitMQConnection(ConnectionFactory factory)
{
_factory = factory ?? throw new ArgumentNullException(nameof(factory));
_connection = factory.CreateConnectionAsync().Result;
}
public async Task<IChannel> CreateChannel()
{
EnsureNotDisposed();
return await _connection.CreateChannelAsync();
}
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
protected virtual void Dispose(bool disposing)
{
if (_isDisposed) return;
if (disposing)
{
// Free any other managed objects here.
}
// Free any unmanaged objects here.
_connection.Dispose();
_isDisposed = true;
}
~RabbitMQConnection()
{
Dispose(false);
}
private void EnsureNotDisposed()
{
if (_isDisposed)
{
throw new ObjectDisposedException(nameof(RabbitMQConnection));
}
}
}
}

13
QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/RabbitMQOptions.cs Normal file
View File

@ -0,0 +1,13 @@
namespace QYZH.InteractiveMagazine.Infrastructure.RabbitMQ
{
public class RabbitMQOptions
{
public string UserName { get; set; }
public string Password { get; set; }
public string HostName { get; set; }
public int Port { get; set; }
public string ExchangeName { get; set; } = string.Empty;
public string VirtualHost { get; set; } = "/";
}
}

25
QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/RabbitMQSendParam.cs Normal file
View File

@ -0,0 +1,25 @@
namespace QYZH.InteractiveMagazine.Infrastructure.RabbitMQ
{
public class RabbitMQSendParam
{
/// <summary>
/// 交换机 默认空
/// </summary>
public string Exchange { get; set; } = "";
public string Queue { get; set; }
/// <summary>
/// 路由键
/// </summary>
public string RoutingKey { get; set; }
/// <summary>
/// 消息数据
/// </summary>
public object Data { get; set; }
/// <summary>
/// 是否清空队列
/// </summary>
public bool Purge { get; set; } = false;
}
}

157
QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/RabbitMQService.cs Normal file
View File

@ -0,0 +1,157 @@
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Encodings.Web;
using System.Text.Json;
namespace QYZH.InteractiveMagazine.Infrastructure.RabbitMQ
{
public class RabbitMQService : IRabbitMQService
{
private readonly IRabbitMQConnection _connection;
private readonly JsonSerializerOptions options = new JsonSerializerOptions
{
Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping
};
public RabbitMQService(IRabbitMQConnection connection)
{
_connection = connection ?? throw new ArgumentNullException(nameof(connection));
}
public async Task<bool> SendAsync(RabbitMQSendParam param, CancellationToken cancellationToken = default)
{
try
{
using var channel = await _connection.CreateChannel();
// 启用发布者确认
//var channelOpts = new CreateChannelOptions
//(
// publisherConfirmationsEnabled: true,
// publisherConfirmationTrackingEnabled: true,
// outstandingPublisherConfirmationsRateLimiter: new ThrottlingRateLimiter(MAX_OUTSTANDING_CONFIRMS)
//);
// 声明队列（持久化）
await channel.QueueDeclareAsync(queue: param.RoutingKey, durable: true, exclusive: false, autoDelete: false, arguments: null);
// 清空队列
if (param.Purge) await channel.QueuePurgeAsync(param.RoutingKey);
// 消息序列化
var mesjson = JsonSerializer.Serialize(param.Data, options);
var body = Encoding.UTF8.GetBytes(mesjson);
var properties = new BasicProperties
{
Persistent = true // 设置消息持久化
};
await channel.BasicPublishAsync(param.Exchange, param.RoutingKey, false, properties, body, cancellationToken);
return true;
}
catch (OperationCanceledException ex)
{
Console.WriteLine($"Operation was canceled: {ex.Message}");
return false;
}
catch (Exception ex)
{
Console.WriteLine($"An error occurred: {ex.Message}");
return false;
}
}
public async Task<bool> SendBatchAsync(IEnumerable<RabbitMQSendParam> @params, CancellationToken cancellationToken = default)
{
IChannel channel = null;
try
{
channel = await _connection.CreateChannel();
// 开启事务
await channel.TxSelectAsync();
var properties = new BasicProperties
{
Persistent = true // 设置消息持久化
};
// 批量发送消息到不同的 routingKey
var declaredExchanges = new HashSet<string>();
foreach (var param in @params)
{
// 声明 Exchange（持久化）
if (declaredExchanges.Add(param.Exchange))
{
await channel.ExchangeDeclareAsync(exchange: param.Exchange, type: "direct", durable: true, autoDelete: false, arguments: null);
}
// 声明队列（持久化）
await channel.QueueDeclareAsync(queue: param.Queue, durable: true, exclusive: false, autoDelete: false, arguments: null);
// 绑定队列到 Exchange
await channel.QueueBindAsync(queue: param.Queue, exchange: param.Exchange, routingKey: param.RoutingKey, arguments: null);
// 清空队列
if (param.Purge) await channel.QueuePurgeAsync(param.Queue);
// 消息序列化
var mesjson = JsonSerializer.Serialize(param.Data, options);
var body = Encoding.UTF8.GetBytes(mesjson);
// 发布消息
await channel.BasicPublishAsync(param.Exchange, param.RoutingKey, false, properties, body, cancellationToken);
}
// 提交事务 - 确保所有消息都发送成功
await channel.TxCommitAsync();
return true;
}
catch (Exception ex)
{
Console.WriteLine($"An error occurred: {ex.Message}");
// 回滚事务
try { await channel?.TxRollbackAsync(); } catch { /* 忽略回滚异常 */ }
return false;
}
finally
{
if (channel != null && !channel.IsClosed)
{
await channel.CloseAsync();
}
}
}
public async Task ReceiveAsync(string queueName, Func<IChannel, BasicDeliverEventArgs, Task> callback, CancellationToken cancellationToken = default)
{
var channel = await _connection.CreateChannel();
//await channel.BasicQosAsync(0, 10, false); // 一次最多接收10条未确认的消息
await channel.QueueDeclareAsync(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
var consumer = new AsyncEventingBasicConsumer(channel);
consumer.ReceivedAsync += async (model, ea) =>
{
//var body = ea.Body.ToArray();
try
{
// 直接传递 model 和 body 给 callback，不需要转换
await callback(channel, ea);
}
finally
{
//await channel.BasicAckAsync(ea.DeliveryTag, false, cancellationToken);
}
};
await channel.BasicConsumeAsync(queue: queueName, autoAck: false, consumer: consumer, cancellationToken: cancellationToken);
// Prevent the method from returning immediately
await Task.Delay(-1, cancellationToken);
}
}
}

56
QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/RabbiteMQExtensions.cs Normal file
View File

@ -0,0 +1,56 @@
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
namespace QYZH.InteractiveMagazine.Infrastructure.RabbitMQ
{
public static class RabbiteMQExtensions
{
/// <summary>
/// 初始化消息队列，并添加Publisher到IoC容器
/// </summary>
/// <remarks>从Configuration读取"RabbbitMQOptions配置项"</remarks>
public static IServiceCollection AddRabbitMQ(this IServiceCollection services, IConfiguration configuration)
{
var rabbitMqSection = configuration.GetSection("RabbitMq");
if (rabbitMqSection.Exists())
{
// 绑定RabbitMQ配置
services.Configure<RabbitMQOptions>(rabbitMqSection);
// 注册RabbitMQ连接工厂
services.AddSingleton<IRabbitMQConnection, RabbitMQConnection>(sp =>
{
var options = sp.GetRequiredService<IOptions<RabbitMQOptions>>().Value;
var factory = new ConnectionFactory()
{
HostName = options.HostName,
Port = options.Port,
UserName = options.UserName,
Password = options.Password,
VirtualHost = options.VirtualHost,
// 自动恢复配置
AutomaticRecoveryEnabled = true, // 启用自动恢复
NetworkRecoveryInterval = TimeSpan.FromSeconds(10), // 每10秒尝试重连
// 心跳检测
RequestedHeartbeat = TimeSpan.FromSeconds(10), // 60秒心跳
// 其他重要配置
TopologyRecoveryEnabled = true, // 恢复交换机、队列等拓扑结构
RequestedConnectionTimeout = TimeSpan.FromSeconds(30), // 连接超时
SocketReadTimeout = TimeSpan.FromSeconds(30), // 读取超时
SocketWriteTimeout = TimeSpan.FromSeconds(30) // 写入超时
};
return new RabbitMQConnection(factory);
});
// 添加RabbitMQService的服务注册
services.AddSingleton<IRabbitMQService, RabbitMQService>();
//services.AddHostedService<TerminalReportService>();
}
return services;
}
}
}