From cbdee5068ac6c22bc20aa1eb9985d6625a7c63e2 Mon Sep 17 00:00:00 2001 From: glz <694770232@qq.com> Date: Fri, 10 Jul 2026 10:44:00 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E6=B6=88=E6=81=AFOut?= =?UTF-8?q?box=E6=9C=BA=E5=88=B6=E3=80=81=E9=9B=AA=E8=8A=B1ID=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E4=BC=98=E5=8C=96=E5=8F=8A=E5=A4=9A=E9=A1=B9=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=E5=AE=8C=E5=96=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 新增数据库唯一约束和Message_Outbox表脚本 2. 新增雪花ID、Hangfire存储、MQ重试等配置实体 3. 重构各项目雪花ID生成逻辑，改为从配置读取WorkerId 4. 优化积分服务分页查询、用户背包更新逻辑 5. 新增JWT令牌Redis过期刷新逻辑 6. 完善RabbitMQ死信队列消息头信息 7. 新增可靠MQ消息发布服务和Outbox派发后台服务 8. 替换原有RabbitMQ直接发送为Outbox可靠发布 9. 优化签到服务逻辑，新增重复签到校验和补签卡扣减逻辑 10. 修复自动铺码消费逻辑，新增点阵页预占和释放机制 --- DatabaseScripts/message_outbox.sql | 22 ++++ .../stability_unique_constraints.sql | 5 + .../IMessagePublishService.cs | 25 ++++ .../DependencyInjectionExtensions.cs | 10 ++ .../Middleware/JwtAutoRefreshMiddleware.cs | 82 +------------ .../RabbitMQ/RabbitMQService.cs | 13 +- .../Dto/RabbitMQ/MessagePublishInput.cs | 38 ++++++ .../Dto/RabbitMQ/MessagePublishResult.cs | 22 ++++ .../Entity/MessageOutbox.cs | 60 ++++++++++ .../Enum/MessageOutboxStatusEnum.cs | 33 ++++++ .../Settings/HangfireStorageSettings.cs | 12 ++ .../Settings/RabbitMQRetrySettings.cs | 17 +++ .../Settings/SnowflakeSettings.cs | 12 ++ .../Consumers/AutoDotCodeConsumer.cs | 43 +++++++ .../Consumers/RabbitMQHostedService.cs | 13 +- .../Program.cs | 4 +- .../appsettings.json | 7 ++ .../BaseRepository.cs | 8 +- .../CheckInService.cs | 56 ++++++--- .../JournalPageService.cs | 26 ++-- .../JournalService.cs | 49 +++++--- .../MessagePublishService.cs | 112 ++++++++++++++++++ .../PointsService.cs | 3 +- .../UserJournalService.cs | 20 ++-- .../WxMallService.cs | 4 +- QYZH.InteractiveMagazine.WeChatApi/Program.cs | 3 +- .../appsettings.json | 3 + QYZH.InteractiveMagazine.WebApi/Program.cs | 4 +- .../appsettings.json | 3 + .../Consumers/RabbitMQHostedService.cs | 13 +- .../Jobs/MessageOutboxDispatchService.cs | 106 +++++++++++++++++ .../Program.cs | 5 +- .../appsettings.json | 12 +- 33 files changed, 693 insertions(+), 152 deletions(-) create mode 100644 DatabaseScripts/message_outbox.sql create mode 100644 DatabaseScripts/stability_unique_constraints.sql create mode 100644 QYZH.InteractiveMagazine.IService/IMessagePublishService.cs create mode 100644 QYZH.InteractiveMagazine.Models/Dto/RabbitMQ/MessagePublishInput.cs create mode 100644 QYZH.InteractiveMagazine.Models/Dto/RabbitMQ/MessagePublishResult.cs create mode 100644 QYZH.InteractiveMagazine.Models/Entity/MessageOutbox.cs create mode 100644 QYZH.InteractiveMagazine.Models/Enum/MessageOutboxStatusEnum.cs create mode 100644 QYZH.InteractiveMagazine.Models/Settings/HangfireStorageSettings.cs create mode 100644 QYZH.InteractiveMagazine.Models/Settings/RabbitMQRetrySettings.cs create mode 100644 QYZH.InteractiveMagazine.Models/Settings/SnowflakeSettings.cs create mode 100644 QYZH.InteractiveMagazine.Service/MessagePublishService.cs create mode 100644 QYZH.InteractiveMagazine.WorkService/Jobs/MessageOutboxDispatchService.cs diff --git a/DatabaseScripts/message_outbox.sql b/DatabaseScripts/message_outbox.sql new file mode 100644 index 0000000..1d7b8bd --- /dev/null +++ b/DatabaseScripts/message_outbox.sql @@ -0,0 +1,22 @@ +CREATE TABLE IF NOT EXISTS `Message_Outbox` ( + `Id` bigint NOT NULL, + `Exchange` varchar(200) NOT NULL, + `Queue` varchar(200) NOT NULL, + `RoutingKey` varchar(200) NOT NULL, + `Payload` json NOT NULL, + `RetryCount` int NOT NULL DEFAULT 0, + `NextRetryAt` datetime NULL, + `SentAt` datetime NULL, + `LastError` varchar(2000) NULL, + `BusinessType` varchar(100) NOT NULL, + `BusinessId` bigint NOT NULL, + `Status` int NOT NULL DEFAULT 0, + `IsDeleted` bit NOT NULL DEFAULT b'0', + `CreatedBy` varchar(100) NOT NULL, + `CreatedAt` datetime NOT NULL, + `UpdatedBy` varchar(100) NULL, + `UpdatedAt` datetime NULL, + PRIMARY KEY (`Id`), + KEY `idx_message_outbox_status_next_retry` (`Status`, `NextRetryAt`, `CreatedAt`), + KEY `idx_message_outbox_business` (`BusinessType`, `BusinessId`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/DatabaseScripts/stability_unique_constraints.sql b/DatabaseScripts/stability_unique_constraints.sql new file mode 100644 index 0000000..7cbad47 --- /dev/null +++ b/DatabaseScripts/stability_unique_constraints.sql @@ -0,0 +1,5 @@ +ALTER TABLE `CheckIn_Record` +ADD UNIQUE KEY `uk_checkin_record_user_date_type_deleted` (`UserId`, `CheckInDate`, `Type`, `IsDeleted`); + +ALTER TABLE `User_Bag` +ADD UNIQUE KEY `uk_user_bag_available_item` (`UserId`, `ItemId`, `Status`, `IsDeleted`); diff --git a/QYZH.InteractiveMagazine.IService/IMessagePublishService.cs b/QYZH.InteractiveMagazine.IService/IMessagePublishService.cs new file mode 100644 index 0000000..e3c20e2 --- /dev/null +++ b/QYZH.InteractiveMagazine.IService/IMessagePublishService.cs @@ -0,0 +1,25 @@ +using QYZH.InteractiveMagazine.Models.Dto.RabbitMQ; +using QYZH.InteractiveMagazine.Models.Entity; + +namespace QYZH.InteractiveMagazine.IService; + +/// +/// MQ消息发布服务。 +/// +public interface IMessagePublishService : IBaseService +{ + /// + /// 可靠发布消息，默认写入Outbox。 + /// + Task PublishAsync(MessagePublishInput input, CancellationToken cancellationToken = default); + + /// + /// 批量可靠发布消息，默认写入Outbox。 + /// + Task PublishBatchAsync(IEnumerable> inputs, CancellationToken cancellationToken = default); + + /// + /// 直接发布消息，不写Outbox。 + /// + Task PublishDirectAsync(MessagePublishInput input, CancellationToken cancellationToken = default); +} diff --git a/QYZH.InteractiveMagazine.Infrastructure/Extensions/DependencyInjectionExtensions.cs b/QYZH.InteractiveMagazine.Infrastructure/Extensions/DependencyInjectionExtensions.cs index 9ad59c6..7122f7f 100644 --- a/QYZH.InteractiveMagazine.Infrastructure/Extensions/DependencyInjectionExtensions.cs +++ b/QYZH.InteractiveMagazine.Infrastructure/Extensions/DependencyInjectionExtensions.cs @@ -106,6 +106,7 @@ public static class DependencyInjectionExtensions var adminToken = await RedisHelper.GetAsync(JwtHelper.BuildAdminTokenKey(userId)); if (adminToken == currentToken) { + await RefreshRedisTokenExpiryAsync(JwtHelper.BuildAdminTokenKey(userId), currentToken, jwtSettings); return; } @@ -116,6 +117,7 @@ public static class DependencyInjectionExtensions var wechatToken = await RedisHelper.GetAsync(JwtHelper.BuildWeChatTokenKey(wxUserId, userId)); if (wechatToken == currentToken) { + await RefreshRedisTokenExpiryAsync(JwtHelper.BuildWeChatTokenKey(wxUserId, userId), currentToken, jwtSettings); return; } @@ -124,6 +126,14 @@ public static class DependencyInjectionExtensions }; } + private static async Task RefreshRedisTokenExpiryAsync(string tokenKey, string currentToken, JwtSettings jwtSettings) + { + if (jwtSettings.ExpiryMinutes > 0) + { + await RedisHelper.SetAsync(tokenKey, currentToken, TimeSpan.FromMinutes(jwtSettings.ExpiryMinutes)); + } + } + private static string? GetBearerToken(TokenValidatedContext context) { var authHeader = context.HttpContext.Request.Headers.Authorization.FirstOrDefault(); diff --git a/QYZH.InteractiveMagazine.Infrastructure/Middleware/JwtAutoRefreshMiddleware.cs b/QYZH.InteractiveMagazine.Infrastructure/Middleware/JwtAutoRefreshMiddleware.cs index fc5aa69..fdcd34a 100644 --- a/QYZH.InteractiveMagazine.Infrastructure/Middleware/JwtAutoRefreshMiddleware.cs +++ b/QYZH.InteractiveMagazine.Infrastructure/Middleware/JwtAutoRefreshMiddleware.cs @@ -1,99 +1,21 @@ using Microsoft.AspNetCore.Http; -using Microsoft.Extensions.Configuration; -using QYZH.InteractiveMagazine.Infrastructure.Auth; -using QYZH.InteractiveMagazine.Models.Settings; -using System.IdentityModel.Tokens.Jwt; -using System.Security.Claims; namespace QYZH.InteractiveMagazine.Infrastructure.Middleware; /// -/// Refreshes the Redis session TTL for the current JWT when it is still the active token. +/// 保留兼容的JWT自动刷新中间件，实际刷新在认证成功后执行。 /// public class JwtAutoRefreshMiddleware { private readonly RequestDelegate _next; - private readonly IConfiguration _configuration; - public JwtAutoRefreshMiddleware(RequestDelegate next, IConfiguration configuration) + public JwtAutoRefreshMiddleware(RequestDelegate next) { _next = next; - _configuration = configuration; } public async Task InvokeAsync(HttpContext context) { - var authHeader = context.Request.Headers.Authorization.FirstOrDefault(); - if (!string.IsNullOrEmpty(authHeader) && authHeader.StartsWith("Bearer ")) - { - var token = authHeader.Substring("Bearer ".Length).Trim(); - await TryRefreshRedisTokenExpiryAsync(token); - } - await _next(context); } - - private async Task TryRefreshRedisTokenExpiryAsync(string token) - { - try - { - var tokenHandler = new JwtSecurityTokenHandler(); - if (tokenHandler.ReadToken(token) is not JwtSecurityToken jwtToken) - { - return; - } - - if (jwtToken.ValidTo <= DateTime.UtcNow) - { - return; - } - - var userId = GetClaimValue(jwtToken, ClaimTypes.NameIdentifier, JwtRegisteredClaimNames.NameId); - if (string.IsNullOrEmpty(userId)) - { - return; - } - - var jwtSettings = _configuration.GetSection("JwtSettings").Get(); - if (jwtSettings == null || jwtSettings.ExpiryMinutes <= 0) - { - return; - } - - var wxUserId = GetClaimValue(jwtToken, JwtHelper.WxUserIdClaimType); - await RefreshRedisTokenExpiryAsync(wxUserId, userId, token, jwtSettings.ExpiryMinutes); - } - catch - { - // Ignore refresh failures. Authentication middleware will validate the request later. - } - } - - private static async Task RefreshRedisTokenExpiryAsync(string? wxUserId, string userId, string currentToken, int expiryMinutes) - { - var adminTokenKey = JwtHelper.BuildAdminTokenKey(userId); - var adminToken = await RedisHelper.GetAsync(adminTokenKey); - if (string.IsNullOrEmpty(wxUserId) && !string.IsNullOrEmpty(adminToken)) - { - await RedisHelper.SetAsync(adminTokenKey, adminToken, TimeSpan.FromMinutes(expiryMinutes)); - return; - } - - if (string.IsNullOrEmpty(wxUserId)) - { - return; - } - - var wechatTokenKey = JwtHelper.BuildWeChatTokenKey(wxUserId, userId); - var wechatToken = await RedisHelper.GetAsync(wechatTokenKey); - if (wechatToken == currentToken) - { - await RedisHelper.SetAsync(wechatTokenKey, currentToken, TimeSpan.FromMinutes(expiryMinutes)); - } - } - - private static string? GetClaimValue(JwtSecurityToken jwtToken, params string[] claimTypes) - { - return jwtToken.Claims.FirstOrDefault(c => claimTypes.Contains(c.Type))?.Value; - } } diff --git a/QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/RabbitMQService.cs b/QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/RabbitMQService.cs index 7d1cbc1..61a2ae6 100644 --- a/QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/RabbitMQService.cs +++ b/QYZH.InteractiveMagazine.Infrastructure/RabbitMQ/RabbitMQService.cs @@ -1,6 +1,7 @@ using RabbitMQ.Client; using RabbitMQ.Client.Events; using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging; using System.Text; using System.Text.Encodings.Web; using System.Text.Json; @@ -11,14 +12,16 @@ namespace QYZH.InteractiveMagazine.Infrastructure.RabbitMQ { private readonly IRabbitMQConnection _connection; private readonly IConfiguration _configuration; + private readonly ILogger _logger; private readonly JsonSerializerOptions options = new JsonSerializerOptions { Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping }; - public RabbitMQService(IRabbitMQConnection connection, IConfiguration configuration) + public RabbitMQService(IRabbitMQConnection connection, IConfiguration configuration, ILogger logger) { _connection = connection ?? throw new ArgumentNullException(nameof(connection)); _configuration = configuration; + _logger = logger; } @@ -55,12 +58,14 @@ namespace QYZH.InteractiveMagazine.Infrastructure.RabbitMQ } catch (OperationCanceledException ex) { - Console.WriteLine($"Operation was canceled: {ex.Message}"); + _logger.LogWarning(ex, "RabbitMQ消息发送已取消，Exchange: {Exchange}, Queue: {Queue}, RoutingKey: {RoutingKey}", + param.Exchange, param.Queue, param.RoutingKey); return false; } catch (Exception ex) { - Console.WriteLine($"An error occurred: {ex.Message}"); + _logger.LogError(ex, "RabbitMQ消息发送失败，Exchange: {Exchange}, Queue: {Queue}, RoutingKey: {RoutingKey}", + param.Exchange, param.Queue, param.RoutingKey); return false; } } @@ -115,7 +120,7 @@ namespace QYZH.InteractiveMagazine.Infrastructure.RabbitMQ } catch (Exception ex) { - Console.WriteLine($"An error occurred: {ex.Message}"); + _logger.LogError(ex, "RabbitMQ批量消息发送失败"); // 回滚事务 try { await channel?.TxRollbackAsync(); } catch { /* 忽略回滚异常 */ } return false; diff --git a/QYZH.InteractiveMagazine.Models/Dto/RabbitMQ/MessagePublishInput.cs b/QYZH.InteractiveMagazine.Models/Dto/RabbitMQ/MessagePublishInput.cs new file mode 100644 index 0000000..174bea4 --- /dev/null +++ b/QYZH.InteractiveMagazine.Models/Dto/RabbitMQ/MessagePublishInput.cs @@ -0,0 +1,38 @@ +namespace QYZH.InteractiveMagazine.Models.Dto.RabbitMQ; + +/// +/// MQ消息发布入参。 +/// +/// 消息数据类型。 +public class MessagePublishInput +{ + /// + /// 交换机。 + /// + public string Exchange { get; set; } = string.Empty; + + /// + /// 队列。 + /// + public string Queue { get; set; } = string.Empty; + + /// + /// 路由键。 + /// + public string RoutingKey { get; set; } = string.Empty; + + /// + /// 消息数据。 + /// + public T Data { get; set; } = default!; + + /// + /// 业务类型。 + /// + public string BusinessType { get; set; } = string.Empty; + + /// + /// 业务ID。 + /// + public long BusinessId { get; set; } +} diff --git a/QYZH.InteractiveMagazine.Models/Dto/RabbitMQ/MessagePublishResult.cs b/QYZH.InteractiveMagazine.Models/Dto/RabbitMQ/MessagePublishResult.cs new file mode 100644 index 0000000..f6b3b58 --- /dev/null +++ b/QYZH.InteractiveMagazine.Models/Dto/RabbitMQ/MessagePublishResult.cs @@ -0,0 +1,22 @@ +namespace QYZH.InteractiveMagazine.Models.Dto.RabbitMQ; + +/// +/// MQ消息发布结果。 +/// +public class MessagePublishResult +{ + /// + /// 是否成功。 + /// + public bool Success { get; set; } + + /// + /// Outbox消息ID。 + /// + public List OutboxIds { get; set; } = []; + + /// + /// 结果消息。 + /// + public string Message { get; set; } = string.Empty; +} diff --git a/QYZH.InteractiveMagazine.Models/Entity/MessageOutbox.cs b/QYZH.InteractiveMagazine.Models/Entity/MessageOutbox.cs new file mode 100644 index 0000000..89819bb --- /dev/null +++ b/QYZH.InteractiveMagazine.Models/Entity/MessageOutbox.cs @@ -0,0 +1,60 @@ +using SqlSugar; + +namespace QYZH.InteractiveMagazine.Models.Entity; + +/// +/// 消息Outbox。 +/// +[SugarTable("Message_Outbox")] +public partial class MessageOutbox : SqlSugarBaseEntity +{ + /// + /// 交换机。 + /// + public string Exchange { get; set; } = string.Empty; + + /// + /// 队列。 + /// + public string Queue { get; set; } = string.Empty; + + /// + /// 路由键。 + /// + public string RoutingKey { get; set; } = string.Empty; + + /// + /// 消息内容JSON。 + /// + public string Payload { get; set; } = string.Empty; + + /// + /// 重试次数。 + /// + public int RetryCount { get; set; } + + /// + /// 下次重试时间。 + /// + public DateTime? NextRetryAt { get; set; } + + /// + /// 发送成功时间。 + /// + public DateTime? SentAt { get; set; } + + /// + /// 最后错误。 + /// + public string? LastError { get; set; } + + /// + /// 业务类型。 + /// + public string BusinessType { get; set; } = string.Empty; + + /// + /// 业务ID。 + /// + public long BusinessId { get; set; } +} diff --git a/QYZH.InteractiveMagazine.Models/Enum/MessageOutboxStatusEnum.cs b/QYZH.InteractiveMagazine.Models/Enum/MessageOutboxStatusEnum.cs new file mode 100644 index 0000000..26deab1 --- /dev/null +++ b/QYZH.InteractiveMagazine.Models/Enum/MessageOutboxStatusEnum.cs @@ -0,0 +1,33 @@ +using System.ComponentModel; + +namespace QYZH.InteractiveMagazine.Models.Enum; + +/// +/// 消息Outbox状态。 +/// +public enum MessageOutboxStatusEnum +{ + /// + /// 待发送。 + /// + [Description("待发送")] + Pending = 0, + + /// + /// 已发送。 + /// + [Description("已发送")] + Sent = 1, + + /// + /// 发送失败待重试。 + /// + [Description("发送失败待重试")] + Failed = 2, + + /// + /// 已放弃。 + /// + [Description("已放弃")] + Abandoned = 3 +} diff --git a/QYZH.InteractiveMagazine.Models/Settings/HangfireStorageSettings.cs b/QYZH.InteractiveMagazine.Models/Settings/HangfireStorageSettings.cs new file mode 100644 index 0000000..8530f75 --- /dev/null +++ b/QYZH.InteractiveMagazine.Models/Settings/HangfireStorageSettings.cs @@ -0,0 +1,12 @@ +namespace QYZH.InteractiveMagazine.Models.Settings; + +/// +/// Hangfire存储配置。 +/// +public class HangfireStorageSettings +{ + /// + /// 存储类型，当前默认 Memory，可配置为 Redis。 + /// + public string StorageType { get; set; } = "Memory"; +} diff --git a/QYZH.InteractiveMagazine.Models/Settings/RabbitMQRetrySettings.cs b/QYZH.InteractiveMagazine.Models/Settings/RabbitMQRetrySettings.cs new file mode 100644 index 0000000..ab2d7ad --- /dev/null +++ b/QYZH.InteractiveMagazine.Models/Settings/RabbitMQRetrySettings.cs @@ -0,0 +1,17 @@ +namespace QYZH.InteractiveMagazine.Models.Settings; + +/// +/// RabbitMQ重试配置。 +/// +public class RabbitMQRetrySettings +{ + /// + /// 最大重试次数。 + /// + public int MaxRetryCount { get; set; } = 3; + + /// + /// 重试延迟毫秒数。 + /// + public int RetryDelayMilliseconds { get; set; } = 30000; +} diff --git a/QYZH.InteractiveMagazine.Models/Settings/SnowflakeSettings.cs b/QYZH.InteractiveMagazine.Models/Settings/SnowflakeSettings.cs new file mode 100644 index 0000000..6f22d49 --- /dev/null +++ b/QYZH.InteractiveMagazine.Models/Settings/SnowflakeSettings.cs @@ -0,0 +1,12 @@ +namespace QYZH.InteractiveMagazine.Models.Settings; + +/// +/// 雪花ID配置。 +/// +public class SnowflakeSettings +{ + /// + /// WorkerId，生产部署时每个写库进程必须唯一。 + /// + public ushort WorkerId { get; set; } +} diff --git a/QYZH.InteractiveMagazine.PrintWorker/Consumers/AutoDotCodeConsumer.cs b/QYZH.InteractiveMagazine.PrintWorker/Consumers/AutoDotCodeConsumer.cs index e785f5d..cd518bb 100644 --- a/QYZH.InteractiveMagazine.PrintWorker/Consumers/AutoDotCodeConsumer.cs +++ b/QYZH.InteractiveMagazine.PrintWorker/Consumers/AutoDotCodeConsumer.cs @@ -48,6 +48,8 @@ public class AutoDotCodeConsumer( using var scope = scopeFactory.CreateScope(); var dbContext = scope.ServiceProvider.GetRequiredService(); JournalPagePrintDto? request = null; + List reservedDotDetailIds = []; + var dotPagesCommitted = false; try { @@ -135,6 +137,21 @@ public class AutoDotCodeConsumer( } var dotFileDetailPageName = dotFileDetailList.Select(x => x.PageName).OrderBy(x => x).ToArray(); + reservedDotDetailIds = dotFileDetailList.Select(x => x.Id).ToList(); + var reservedRows = await dbContext.Updateable() + .SetColumns(x => x.IsUse == true) + .SetColumns(x => x.UpdatedAt == DateTime.Now) + .Where(x => reservedDotDetailIds.Contains(x.Id) && !x.IsUse) + .ExecuteCommandAsync(cancellationToken); + if (reservedRows != reservedDotDetailIds.Count) + { + logger.LogError("点阵页码预占失败，DotId: {DotId}, Need: {Need}, Reserved: {Reserved}", dotId, reservedDotDetailIds.Count, reservedRows); + await ReleaseReservedDotPagesAsync(dbContext, reservedDotDetailIds, cancellationToken); + reservedDotDetailIds.Clear(); + await TryCallbackCodeFailAsync(request, cancellationToken, dotFileDetailPageName); + return; + } + var pageStr = "{" + $"[{dotFileDetailList[0].PageName},{pageNumMax}]" + "}"; var arguments = $"-sMode=Generate -sPDF=\"{uploadFilePath}\" -sLIC=\"{xmlPath}\" -pStart=1 -oPDF=\"{downloadFilePath}\" -dPageAddr=1 -dPrint={dPrint} -dDotSize=40 -dType=0 -dOutFile=0 -dControlPageNum={pageStr}"; var printResult = await ExecutePrintToolAsync(exePath, printToolDirectory, arguments, cancellationToken); @@ -142,6 +159,8 @@ public class AutoDotCodeConsumer( if (printResult.Timeout || printResult.ExitCode != 0) { logger.LogError("执行 PrintTool.exe 失败，退出码：{ExitCode}，错误信息：{ErrorMessage}", printResult.ExitCode, printResult.Error); + await ReleaseReservedDotPagesAsync(dbContext, reservedDotDetailIds, cancellationToken); + reservedDotDetailIds.Clear(); await TryCallbackCodeFailAsync(request, cancellationToken, dotFileDetailPageName); return; } @@ -154,6 +173,8 @@ public class AutoDotCodeConsumer( if (!ValidatePrintOutput(printResult.Output, dotFileDetailPageName, dPrint)) { + await ReleaseReservedDotPagesAsync(dbContext, reservedDotDetailIds, cancellationToken); + reservedDotDetailIds.Clear(); await TryCallbackCodeFailAsync(request, cancellationToken, dotFileDetailPageName); return; } @@ -189,9 +210,12 @@ public class AutoDotCodeConsumer( .Where(x => x.Id == dotId) .ExecuteCommandAsync(); }); + dotPagesCommitted = true; } else { + await ReleaseReservedDotPagesAsync(dbContext, reservedDotDetailIds, cancellationToken); + reservedDotDetailIds.Clear(); logger.LogError("回调接口修改书籍状态失败，JournalId: {JournalId}，接口返回消息：{Message}", statusModel.JournalId, callbackResponse.Message); } } @@ -205,6 +229,11 @@ public class AutoDotCodeConsumer( logger.LogError(ex, "自动铺码处理失败"); if (request != null) { + if (!dotPagesCommitted && reservedDotDetailIds.Count > 0) + { + await ReleaseReservedDotPagesAsync(dbContext, reservedDotDetailIds, CancellationToken.None); + } + await TryCallbackCodeFailAsync(request, CancellationToken.None); } } @@ -329,6 +358,20 @@ public class AutoDotCodeConsumer( } } + private static async Task ReleaseReservedDotPagesAsync(ISqlSugarClient dbContext, List dotDetailIds, CancellationToken cancellationToken) + { + if (dotDetailIds.Count == 0) + { + return; + } + + await dbContext.Updateable() + .SetColumns(x => x.IsUse == false) + .SetColumns(x => x.UpdatedAt == DateTime.Now) + .Where(x => dotDetailIds.Contains(x.Id)) + .ExecuteCommandAsync(cancellationToken); + } + private static JournalPagePrintDto BuildFailResponse(JournalPagePrintDto request, string[]? pageNo = null) { return new JournalPagePrintDto diff --git a/QYZH.InteractiveMagazine.PrintWorker/Consumers/RabbitMQHostedService.cs b/QYZH.InteractiveMagazine.PrintWorker/Consumers/RabbitMQHostedService.cs index 721dbdc..f4d4505 100644 --- a/QYZH.InteractiveMagazine.PrintWorker/Consumers/RabbitMQHostedService.cs +++ b/QYZH.InteractiveMagazine.PrintWorker/Consumers/RabbitMQHostedService.cs @@ -55,7 +55,7 @@ public class RabbitMQHostedService(IServiceProvider serviceProvider, ILogger SendToDeadLetterQueueAsync(string exchange, string dlqName, string dlqRoutingKey, byte[] body, CancellationToken cancellationToken) + private async Task SendToDeadLetterQueueAsync(string exchange, string dlqName, string dlqRoutingKey, string originalQueueName, byte[] body, Exception exception, CancellationToken cancellationToken) { try { @@ -90,7 +90,14 @@ public class RabbitMQHostedService(IServiceProvider serviceProvider, ILogger + { + ["x-original-queue"] = originalQueueName, + ["x-error-type"] = exception.GetType().FullName, + ["x-error-message"] = exception.Message, + ["x-failed-at"] = DateTimeOffset.UtcNow.ToString("O") + } }; await channel.BasicPublishAsync(exchange, dlqRoutingKey, false, properties, body, cancellationToken); logger.LogInformation("消息已发送到死信队列: {DlqName}", dlqName); diff --git a/QYZH.InteractiveMagazine.PrintWorker/Program.cs b/QYZH.InteractiveMagazine.PrintWorker/Program.cs index 9f9ed11..0adbe73 100644 --- a/QYZH.InteractiveMagazine.PrintWorker/Program.cs +++ b/QYZH.InteractiveMagazine.PrintWorker/Program.cs @@ -1,6 +1,7 @@ using QYZH.InteractiveMagazine.Infrastructure.RabbitMQ; using QYZH.InteractiveMagazine.Infrastructure.Redis; using QYZH.InteractiveMagazine.Infrastructure.SDK; +using QYZH.InteractiveMagazine.Models.Settings; using QYZH.InteractiveMagazine.PrintWorker.Consumers; using Serilog; using SqlSugar; @@ -34,7 +35,8 @@ Log.Logger = new LoggerConfiguration() builder.Services.AddSerilog(); builder.Services.AddWindowsService(options => options.ServiceName = serviceName); -YitIdHelper.SetIdGenerator(new IdGeneratorOptions { WorkerId = 3 }); +var snowflakeSettings = builder.Configuration.GetSection("SnowflakeSettings").Get() ?? new SnowflakeSettings { WorkerId = 3 }; +YitIdHelper.SetIdGenerator(new IdGeneratorOptions { WorkerId = snowflakeSettings.WorkerId }); builder.Services.AddSqlSugar(new IocConfig { diff --git a/QYZH.InteractiveMagazine.PrintWorker/appsettings.json b/QYZH.InteractiveMagazine.PrintWorker/appsettings.json index 028a828..bd31cac 100644 --- a/QYZH.InteractiveMagazine.PrintWorker/appsettings.json +++ b/QYZH.InteractiveMagazine.PrintWorker/appsettings.json @@ -2,6 +2,9 @@ "ConnectionStrings": { "DefaultConnection": "server=192.168.20.150;port=13306;database=InteractiveMagazine;user=user;password=n68792bu!y99r905;charset=utf8mb4;" }, + "SnowflakeSettings": { + "WorkerId": 3 + }, "RedisSettings": { "ConnectionString": "192.168.20.150:16379,defaultDatabase=5", "Sentinels": [], @@ -14,6 +17,10 @@ "Password": "@ss%&*otz%d*pq2S", "VirtualHost": "InteractiveMagazine" }, + "RabbitMQRetrySettings": { + "MaxRetryCount": 3, + "RetryDelayMilliseconds": 30000 + }, "Serilog": { "MinimumLevel": { "Default": "Information", diff --git a/QYZH.InteractiveMagazine.Repository/BaseRepository.cs b/QYZH.InteractiveMagazine.Repository/BaseRepository.cs index d1fb61e..86b38b6 100644 --- a/QYZH.InteractiveMagazine.Repository/BaseRepository.cs +++ b/QYZH.InteractiveMagazine.Repository/BaseRepository.cs @@ -253,9 +253,9 @@ namespace QYZH.InteractiveMagazine.Repository var total = 0; page.PageSize = parm.PageSize; page.PageIndex = parm.PageIndex; - if (string.IsNullOrEmpty(parm.Sort)) + if (!string.IsNullOrWhiteSpace(parm.Sort)) { - source.OrderByPropertyName(parm.Sort, parm.SortType.Contains("desc") ? OrderByType.Desc : OrderByType.Asc); + source.OrderByPropertyName(parm.Sort, parm.SortType?.Contains("desc", StringComparison.OrdinalIgnoreCase) == true ? OrderByType.Desc : OrderByType.Asc); } page.Result = source //.OrderByIF(parm.Sort.IsNotEmpty(), $"{parm.Sort.ToSqlFilter()} {(!string.IsNullOrWhiteSpace(parm.SortType) && parm.SortType.Contains("desc") ? "desc" : "asc")}") @@ -278,9 +278,9 @@ namespace QYZH.InteractiveMagazine.Repository var total = 0; page.PageSize = parm.PageSize; page.PageIndex = parm.PageIndex; - if (string.IsNullOrEmpty(parm.Sort)) + if (!string.IsNullOrWhiteSpace(parm.Sort)) { - source.OrderByPropertyName(parm.Sort, parm.SortType.Contains("desc") ? OrderByType.Desc : OrderByType.Asc); + source.OrderByPropertyName(parm.Sort, parm.SortType?.Contains("desc", StringComparison.OrdinalIgnoreCase) == true ? OrderByType.Desc : OrderByType.Asc); } var result = source //.OrderByIF(parm.Sort.IsNotEmpty(), $"{parm.Sort.ToSqlFilter()} {(!string.IsNullOrWhiteSpace(parm.SortType) && parm.SortType.Contains("desc") ? "desc" : "asc")}") diff --git a/QYZH.InteractiveMagazine.Service/CheckInService.cs b/QYZH.InteractiveMagazine.Service/CheckInService.cs index f967216..34b854f 100644 --- a/QYZH.InteractiveMagazine.Service/CheckInService.cs +++ b/QYZH.InteractiveMagazine.Service/CheckInService.cs @@ -74,6 +74,14 @@ public class CheckInService( await checkInRecordRepository.UseTranAsync(async () => { + var duplicate = await checkInRecordRepository.Context.Queryable() + .Where(r => r.UserId == userId && !r.IsDeleted && r.CheckInDate >= today && r.CheckInDate < today.AddDays(1)) + .AnyAsync(); + if (duplicate) + { + throw new BusinessException("今日已签到，请明天再来", ResultCode.CONFLICT); + } + // 6a. 创建签到记录 var checkInRecord = new CheckInRecord { @@ -94,10 +102,16 @@ public class CheckInService( var newGrowthBalance = user.GrowthPoints + growthReward; await checkInRecordRepository.Context.Updateable() - .SetColumns(u => u.GrowthPoints == newGrowthBalance) + .SetColumns(u => u.GrowthPoints == u.GrowthPoints + growthReward) + .SetColumns(u => u.UpdatedAt == DateTime.Now) .Where(u => u.Id == userId && !u.IsDeleted) .ExecuteCommandAsync(); + newGrowthBalance = await checkInRecordRepository.Context.Queryable() + .Where(u => u.Id == userId && !u.IsDeleted) + .Select(u => u.GrowthPoints) + .FirstAsync(); + // 6c. 通过积分服务增加积分 var pointsResult = await pointsService.AddPointsInTranAsync(new AddPointsInput { @@ -271,6 +285,14 @@ public class CheckInService( await checkInRecordRepository.UseTranAsync(async () => { + var duplicate = await checkInRecordRepository.Context.Queryable() + .Where(r => r.UserId == userId && !r.IsDeleted && r.CheckInDate >= targetDate && r.CheckInDate < targetDate.AddDays(1)) + .AnyAsync(); + if (duplicate) + { + throw new BusinessException($"{targetDate:yyyy-MM-dd} 已签到，无需补签", ResultCode.CONFLICT); + } + // 创建补签记录 var checkInRecord = new CheckInRecord { @@ -290,14 +312,17 @@ public class CheckInService( var recordId = await checkInRecordRepository.Insertable(checkInRecord).ExecuteReturnIdentityAsync(); checkInRecord.Id = recordId; - // 更新用户成长值 - var newGrowthBalance = user.GrowthPoints + growthReward; - await checkInRecordRepository.Context.Updateable() - .SetColumns(u => u.GrowthPoints == newGrowthBalance) + .SetColumns(u => u.GrowthPoints == u.GrowthPoints + growthReward) + .SetColumns(u => u.UpdatedAt == DateTime.Now) .Where(u => u.Id == userId && !u.IsDeleted) .ExecuteCommandAsync(); + var newGrowthBalance = await checkInRecordRepository.Context.Queryable() + .Where(u => u.Id == userId && !u.IsDeleted) + .Select(u => u.GrowthPoints) + .FirstAsync(); + // 通过积分服务增加积分 var pointsResult = await pointsService.AddPointsInTranAsync(new AddPointsInput { @@ -311,19 +336,20 @@ public class CheckInService( // 扣减补签卡 if (makeUpCard.Quantity <= 1) { + var cardAffectedRows = await checkInRecordRepository.Context.Updateable() + .SetColumns(b => b.Quantity == b.Quantity - 1) + .SetColumns(b => b.UpdatedAt == DateTime.Now) + .Where(b => b.Id == makeUpCard.Id && b.UserId == userId && b.Quantity > 0 && b.Status == (int)UserBagStatusEnum.Available && !b.IsDeleted) + .ExecuteCommandAsync(); + if (cardAffectedRows <= 0) + { + throw new BusinessException("补签卡不足，无法补签", ResultCode.CONFLICT); + } + await checkInRecordRepository.Context.Updateable() .SetColumns(b => b.Status == (int)UserBagStatusEnum.Expired) - .SetColumns(b => b.Quantity == 0) .SetColumns(b => b.UpdatedAt == DateTime.Now) - .Where(b => b.Id == makeUpCard.Id) - .ExecuteCommandAsync(); - } - else - { - await checkInRecordRepository.Context.Updateable() - .SetColumns(b => b.Quantity == makeUpCard.Quantity - 1) - .SetColumns(b => b.UpdatedAt == DateTime.Now) - .Where(b => b.Id == makeUpCard.Id) + .Where(b => b.Id == makeUpCard.Id && b.Quantity <= 0 && !b.IsDeleted) .ExecuteCommandAsync(); } diff --git a/QYZH.InteractiveMagazine.Service/JournalPageService.cs b/QYZH.InteractiveMagazine.Service/JournalPageService.cs index cbcaf57..6111da3 100644 --- a/QYZH.InteractiveMagazine.Service/JournalPageService.cs +++ b/QYZH.InteractiveMagazine.Service/JournalPageService.cs @@ -11,6 +11,7 @@ using QYZH.InteractiveMagazine.IService; using QYZH.InteractiveMagazine.Models.Common; using QYZH.InteractiveMagazine.Models.Dto; using QYZH.InteractiveMagazine.Models.Dto.Journal; +using QYZH.InteractiveMagazine.Models.Dto.RabbitMQ; using QYZH.InteractiveMagazine.Models.Entity; using QYZH.InteractiveMagazine.Models.Enum; using QYZH.InteractiveMagazine.Repository; @@ -24,7 +25,7 @@ public class JournalPageService(BaseRepository JournalRepository, OssService ossService, IHttpClientFactory httpClientFactory, IConfiguration configuration, - IRabbitMQService rabbitMqService, + IMessagePublishService messagePublishService, ILogger logger, BaseRepository JournalPageRepository, BaseRepository JournalPageTaskRepository, @@ -135,18 +136,29 @@ public class JournalPageService(BaseRepository JournalRepository, var isPdfExists = ossService.DoesObjectExist(uploadPdfKey); BusinessException.ThrowIf(!isPdfExists, "获取期刊上传的PDF文件失败，请检查PDF文件是否上传成功", ResultCode.GLOBAL_ERROR); - journalEntity.Status = (int)JournalStatusEnum.Codeing; - - await JournalRepository.Updateable(journalEntity).UpdateColumns(x => new { x.Status, x.UpdatedAt }).ExecuteCommandAsync(); - var data = new JournalPagePrintDto { JournalId = JournalId, JournalPdfUrl = uploadPdfUrl, PageNum = [.. journalPageList.Select(x => x.PageNum)] }; - var msRes = await rabbitMqService.SendAsync(new RabbitMQSendParam { Exchange = "ex.journal", Queue = "mq.journal.dotcode.auto", RoutingKey = "rk.journal.dotcode.auto", Data = data });//发生消息 - return msRes; + + return await UseTranAsync(async () => + { + journalEntity.Status = (int)JournalStatusEnum.Codeing; + journalEntity.UpdatedAt = DateTime.Now; + await JournalRepository.Updateable(journalEntity).UpdateColumns(x => new { x.Status, x.UpdatedAt }).ExecuteCommandAsync(); + await messagePublishService.PublishAsync(new MessagePublishInput + { + Exchange = "ex.journal", + Queue = "mq.journal.dotcode.auto", + RoutingKey = "rk.journal.dotcode.auto", + Data = data, + BusinessType = "JournalDotCode", + BusinessId = JournalId + }); + return true; + }); } /// /// 回调接口-自动铺码, 书页铺码后回调接口，更新书页的点阵码 diff --git a/QYZH.InteractiveMagazine.Service/JournalService.cs b/QYZH.InteractiveMagazine.Service/JournalService.cs index 32fbea3..54a5a16 100644 --- a/QYZH.InteractiveMagazine.Service/JournalService.cs +++ b/QYZH.InteractiveMagazine.Service/JournalService.cs @@ -9,10 +9,12 @@ using QYZH.InteractiveMagazine.Models.Common; using QYZH.InteractiveMagazine.Models.Dto; using QYZH.InteractiveMagazine.Models.Dto.DotMatrix; using QYZH.InteractiveMagazine.Models.Dto.Journal; +using QYZH.InteractiveMagazine.Models.Dto.RabbitMQ; using QYZH.InteractiveMagazine.Models.Entity; using QYZH.InteractiveMagazine.Models.Enum; using QYZH.InteractiveMagazine.Repository; using SqlSugar; +using System.Text.Json; using Yitter.IdGenerator; namespace QYZH.InteractiveMagazine.Service; @@ -24,7 +26,7 @@ public class JournalService(BaseRepository JournalPageRepository, BaseRepository dotFileRepository, BaseRepository dotFileDetailRepository, OssService ossService, - IRabbitMQService rabbitMqService, + IMessagePublishService messagePublishService, ILogger logger) : BaseRepository, IJournalService { private const string JournalExchange = "ex.journal"; @@ -375,23 +377,27 @@ public class JournalService(BaseRepository JournalPageRepository, x => x.Key, x => string.Join(',', x.OrderBy(q => ParseTaskNo(q.No)).Select(q => q.Id))); - var bookMessageSent = await rabbitMqService.SendAsync(new RabbitMQSendParam + var publishMessages = new List> { - Exchange = JournalExchange, - Queue = PublishBookQueue, - RoutingKey = PublishBookRoutingKey, - Data = new JournalPublishBookMessage + new() { - BookId = id, - StartTime = book.StartTime.Value, - EndTime = book.EndTime.Value + Exchange = JournalExchange, + Queue = PublishBookQueue, + RoutingKey = PublishBookRoutingKey, + Data = new JournalPublishBookMessage + { + BookId = id, + StartTime = book.StartTime.Value, + EndTime = book.EndTime.Value + }, + BusinessType = "JournalPublish", + BusinessId = id } - }); - BusinessException.ThrowIf(!bookMessageSent, "发布书籍消息发送失败", ResultCode.GLOBAL_ERROR); + }; foreach (var page in pages) { - var pageMessageSent = await rabbitMqService.SendAsync(new RabbitMQSendParam + publishMessages.Add(new MessagePublishInput { Exchange = JournalExchange, Queue = PublishBookPageQueue, @@ -404,16 +410,21 @@ public class JournalService(BaseRepository JournalPageRepository, Layout = page.Layout, Url = DomainHelper.OssFullUrl(page.Url), QuestionNo = questionIdsByPageId.GetValueOrDefault(page.Id) ?? string.Empty - } + }, + BusinessType = "JournalPublish", + BusinessId = id }); - BusinessException.ThrowIf(!pageMessageSent, $"发布书页消息发送失败，PageId: {page.Id}", ResultCode.GLOBAL_ERROR); } - return await base.Updateable() - .SetColumns(s => s.Status, JournalStatusEnum.Published) - .SetColumns(s => s.UpdatedAt, DateTime.Now) - .Where(w => w.Id == id) - .ExecuteCommandAsync() > 0; + return await UseTranAsync(async () => + { + await messagePublishService.PublishBatchAsync(publishMessages); + return await base.Updateable() + .SetColumns(s => s.Status, JournalStatusEnum.Published) + .SetColumns(s => s.UpdatedAt, DateTime.Now) + .Where(w => w.Id == id) + .ExecuteCommandAsync() > 0; + }); static (int First, int Second, int Third, int Fourth) ParseTaskNo(string? no) { diff --git a/QYZH.InteractiveMagazine.Service/MessagePublishService.cs b/QYZH.InteractiveMagazine.Service/MessagePublishService.cs new file mode 100644 index 0000000..a86a260 --- /dev/null +++ b/QYZH.InteractiveMagazine.Service/MessagePublishService.cs @@ -0,0 +1,112 @@ +using Microsoft.Extensions.Logging; +using QYZH.InteractiveMagazine.Infrastructure.RabbitMQ; +using QYZH.InteractiveMagazine.IService; +using QYZH.InteractiveMagazine.Models.Common; +using QYZH.InteractiveMagazine.Models.Dto; +using QYZH.InteractiveMagazine.Models.Dto.RabbitMQ; +using QYZH.InteractiveMagazine.Models.Entity; +using QYZH.InteractiveMagazine.Models.Enum; +using QYZH.InteractiveMagazine.Repository; +using System.Text.Json; + +namespace QYZH.InteractiveMagazine.Service; + +/// +/// MQ消息发布服务。 +/// +public class MessagePublishService( + IRabbitMQService rabbitMQService, + ILogger logger) : BaseRepository, IMessagePublishService +{ + /// + /// 可靠发布消息，默认写入Outbox。 + /// + public async Task PublishAsync(MessagePublishInput input, CancellationToken cancellationToken = default) + { + ValidateInput(input); + var message = BuildOutboxMessage(input); + await Context.Insertable(message).ExecuteCommandAsync(cancellationToken); + return new MessagePublishResult + { + Success = true, + OutboxIds = [message.Id], + Message = "消息已写入Outbox" + }; + } + + /// + /// 批量可靠发布消息，默认写入Outbox。 + /// + public async Task PublishBatchAsync(IEnumerable> inputs, CancellationToken cancellationToken = default) + { + var inputList = inputs.ToList(); + BusinessException.ThrowIf(inputList.Count == 0, "消息发布列表不能为空", ResultCode.BAD_REQUEST); + inputList.ForEach(ValidateInput); + + var messages = inputList.Select(BuildOutboxMessage).ToList(); + await Context.Insertable(messages).ExecuteCommandAsync(cancellationToken); + return new MessagePublishResult + { + Success = true, + OutboxIds = messages.Select(x => x.Id).ToList(), + Message = "消息已批量写入Outbox" + }; + } + + /// + /// 直接发布消息，不写Outbox。 + /// + public async Task PublishDirectAsync(MessagePublishInput input, CancellationToken cancellationToken = default) + { + ValidateInput(input); + var sent = await rabbitMQService.SendAsync(new RabbitMQSendParam + { + Exchange = input.Exchange, + Queue = input.Queue, + RoutingKey = input.RoutingKey, + Data = input.Data! + }, cancellationToken); + + if (!sent) + { + logger.LogError("MQ直接发布失败，Exchange: {Exchange}, Queue: {Queue}, RoutingKey: {RoutingKey}, BusinessType: {BusinessType}, BusinessId: {BusinessId}", + input.Exchange, input.Queue, input.RoutingKey, input.BusinessType, input.BusinessId); + throw new BusinessException("MQ消息发送失败", ResultCode.GLOBAL_ERROR); + } + + return new MessagePublishResult + { + Success = true, + Message = "消息已直接发送" + }; + } + + private static MessageOutbox BuildOutboxMessage(MessagePublishInput input) + { + var now = DateTime.Now; + return new MessageOutbox + { + Exchange = input.Exchange, + Queue = input.Queue, + RoutingKey = input.RoutingKey, + Payload = JsonSerializer.Serialize(input.Data), + BusinessType = input.BusinessType, + BusinessId = input.BusinessId, + Status = (int)MessageOutboxStatusEnum.Pending, + CreatedBy = "System", + CreatedAt = now, + UpdatedBy = "System", + UpdatedAt = now + }; + } + + private static void ValidateInput(MessagePublishInput input) + { + BusinessException.ThrowIf(string.IsNullOrWhiteSpace(input.Exchange), "MQ交换机不能为空", ResultCode.BAD_REQUEST); + BusinessException.ThrowIf(string.IsNullOrWhiteSpace(input.Queue), "MQ队列不能为空", ResultCode.BAD_REQUEST); + BusinessException.ThrowIf(string.IsNullOrWhiteSpace(input.RoutingKey), "MQ路由键不能为空", ResultCode.BAD_REQUEST); + BusinessException.ThrowIf(string.IsNullOrWhiteSpace(input.BusinessType), "MQ业务类型不能为空", ResultCode.BAD_REQUEST); + BusinessException.ThrowIf(input.BusinessId <= 0, "MQ业务ID无效", ResultCode.BAD_REQUEST); + BusinessException.ThrowIf(input.Data == null, "MQ消息数据不能为空", ResultCode.BAD_REQUEST); + } +} diff --git a/QYZH.InteractiveMagazine.Service/PointsService.cs b/QYZH.InteractiveMagazine.Service/PointsService.cs index 53260e2..a0487d3 100644 --- a/QYZH.InteractiveMagazine.Service/PointsService.cs +++ b/QYZH.InteractiveMagazine.Service/PointsService.cs @@ -6,6 +6,7 @@ using QYZH.InteractiveMagazine.Models.Dto.Points; using QYZH.InteractiveMagazine.Models.Entity; using QYZH.InteractiveMagazine.Models.Enum; using QYZH.InteractiveMagazine.Repository; +using SqlSugar; namespace QYZH.InteractiveMagazine.Service; @@ -265,7 +266,7 @@ public class PointsService( .WhereIF(!string.IsNullOrEmpty(input.Status), r => r.Status.ToString() == input.Status) .OrderByDescending(r => r.CreatedAt); - var total = 0; + RefAsync total = 0; var records = await query .Select(r => new PointsRecordOutput { diff --git a/QYZH.InteractiveMagazine.Service/UserJournalService.cs b/QYZH.InteractiveMagazine.Service/UserJournalService.cs index a1085ef..89e2140 100644 --- a/QYZH.InteractiveMagazine.Service/UserJournalService.cs +++ b/QYZH.InteractiveMagazine.Service/UserJournalService.cs @@ -5,6 +5,8 @@ using QYZH.InteractiveMagazine.Infrastructure.RabbitMQ; using QYZH.InteractiveMagazine.IService; using QYZH.InteractiveMagazine.Models.Common; using QYZH.InteractiveMagazine.Models.Dto; +using QYZH.InteractiveMagazine.Models.Dto.Journal; +using QYZH.InteractiveMagazine.Models.Dto.RabbitMQ; using QYZH.InteractiveMagazine.Models.Entity; using QYZH.InteractiveMagazine.Models.Enum; using QYZH.InteractiveMagazine.Repository; @@ -21,7 +23,7 @@ public class UserJournalService( BaseRepository usersRepository, BaseRepository journalRepository, ILogger logger, - IRabbitMQService rabbitMqService, + IMessagePublishService messagePublishService, OssService ossService, IPetService petService) : BaseRepository, IUserJournalService @@ -214,7 +216,7 @@ public class UserJournalService( } var recordIds = records.Select(r => r.Id).ToList(); - var messageSent = await rabbitMqService.SendAsync(new RabbitMQSendParam + var messageSent = (await messagePublishService.PublishAsync(new MessagePublishInput { Exchange = JournalExchange, Queue = QrCodeGenerateQueue, @@ -223,8 +225,10 @@ public class UserJournalService( { RecordIds = recordIds, OperatorName = operatorName - } - }); + }, + BusinessType = "UserJournalQrCodeGenerate", + BusinessId = input.JournalId + })).Success; if (!messageSent) { @@ -457,7 +461,7 @@ public class UserJournalService( { try { - var messageSent = await rabbitMqService.SendAsync(new RabbitMQSendParam + var messageSent = (await messagePublishService.PublishAsync(new MessagePublishInput { Exchange = JournalExchange, Queue = BindJournalQueue, @@ -469,8 +473,10 @@ public class UserJournalService( StartTime = journal.StartTime, EndTime = journal.EndTime, UploadDomain = user.UploadDomain - } - }); + }, + BusinessType = "UserJournalBind", + BusinessId = user.Id + })).Success; if (!messageSent) { diff --git a/QYZH.InteractiveMagazine.Service/WxMallService.cs b/QYZH.InteractiveMagazine.Service/WxMallService.cs index bf0b5eb..606646c 100644 --- a/QYZH.InteractiveMagazine.Service/WxMallService.cs +++ b/QYZH.InteractiveMagazine.Service/WxMallService.cs @@ -253,9 +253,9 @@ public class WxMallService( { // 已有同类物品，累加数量 await exchangeRecordRepository.Context.Updateable() - .SetColumns(b => b.Quantity == existingBag.Quantity + input.Quantity) + .SetColumns(b => b.Quantity == b.Quantity + input.Quantity) .SetColumns(b => b.UpdatedAt == DateTime.Now) - .Where(b => b.Id == existingBag.Id) + .Where(b => b.Id == existingBag.Id && !b.IsDeleted && b.Status == (int)UserBagStatusEnum.Available) .ExecuteCommandAsync(); } else diff --git a/QYZH.InteractiveMagazine.WeChatApi/Program.cs b/QYZH.InteractiveMagazine.WeChatApi/Program.cs index c3dd3bd..70258f6 100644 --- a/QYZH.InteractiveMagazine.WeChatApi/Program.cs +++ b/QYZH.InteractiveMagazine.WeChatApi/Program.cs @@ -26,7 +26,8 @@ var builder = WebApplication.CreateBuilder(args); builder.Configuration.AddJsonFile("medal-rule-config.json", optional: false, reloadOnChange: true); -YitIdHelper.SetIdGenerator(new IdGeneratorOptions { WorkerId = 1 }); +var snowflakeSettings = builder.Configuration.GetSection("SnowflakeSettings").Get() ?? new SnowflakeSettings { WorkerId = 1 }; +YitIdHelper.SetIdGenerator(new IdGeneratorOptions { WorkerId = snowflakeSettings.WorkerId }); builder.UseAutofac(); diff --git a/QYZH.InteractiveMagazine.WeChatApi/appsettings.json b/QYZH.InteractiveMagazine.WeChatApi/appsettings.json index 87c9381..2daedce 100644 --- a/QYZH.InteractiveMagazine.WeChatApi/appsettings.json +++ b/QYZH.InteractiveMagazine.WeChatApi/appsettings.json @@ -9,6 +9,9 @@ "ExpiryMinutes": 120, "JwtTokenExpiryDays": 30 }, + "SnowflakeSettings": { + "WorkerId": 1 + }, "RedisSettings": { "ConnectionString": "192.168.20.150:16379,defaultDatabase=5", "Sentinels": [], diff --git a/QYZH.InteractiveMagazine.WebApi/Program.cs b/QYZH.InteractiveMagazine.WebApi/Program.cs index a06e381..99cf60e 100644 --- a/QYZH.InteractiveMagazine.WebApi/Program.cs +++ b/QYZH.InteractiveMagazine.WebApi/Program.cs @@ -32,8 +32,8 @@ var builder = WebApplication.CreateBuilder(args); // 加载勋章规则独立配置文件 builder.Configuration.AddJsonFile("medal-rule-config.json", optional: false, reloadOnChange: true); -// 初始化雪花ID生成器 -YitIdHelper.SetIdGenerator(new IdGeneratorOptions() { WorkerId = 1 }); +var snowflakeSettings = builder.Configuration.GetSection("SnowflakeSettings").Get() ?? new SnowflakeSettings { WorkerId = 1 }; +YitIdHelper.SetIdGenerator(new IdGeneratorOptions { WorkerId = snowflakeSettings.WorkerId }); // autofac注入 允许使用autofac作为DI容器 builder.UseAutofac(); diff --git a/QYZH.InteractiveMagazine.WebApi/appsettings.json b/QYZH.InteractiveMagazine.WebApi/appsettings.json index 87c9381..2daedce 100644 --- a/QYZH.InteractiveMagazine.WebApi/appsettings.json +++ b/QYZH.InteractiveMagazine.WebApi/appsettings.json @@ -9,6 +9,9 @@ "ExpiryMinutes": 120, "JwtTokenExpiryDays": 30 }, + "SnowflakeSettings": { + "WorkerId": 1 + }, "RedisSettings": { "ConnectionString": "192.168.20.150:16379,defaultDatabase=5", "Sentinels": [], diff --git a/QYZH.InteractiveMagazine.WorkService/Consumers/RabbitMQHostedService.cs b/QYZH.InteractiveMagazine.WorkService/Consumers/RabbitMQHostedService.cs index 12ec068..64aad9d 100644 --- a/QYZH.InteractiveMagazine.WorkService/Consumers/RabbitMQHostedService.cs +++ b/QYZH.InteractiveMagazine.WorkService/Consumers/RabbitMQHostedService.cs @@ -68,7 +68,7 @@ public class RabbitMQHostedService : BackgroundService _logger.LogError(ex, "消费者 {QueueName} 处理消息异常", queueName); // 发送到死信队列，成功则从主队列移除，失败则重新入队 - var dlqSent = await SendToDeadLetterQueueAsync(exchange, dlqName, dlqRoutingKey, ea.Body.ToArray(), stoppingToken); + var dlqSent = await SendToDeadLetterQueueAsync(exchange, dlqName, dlqRoutingKey, queueName, ea.Body.ToArray(), ex, stoppingToken); await channel.BasicNackAsync(ea.DeliveryTag, false, !dlqSent, stoppingToken); if (dlqSent) @@ -100,7 +100,7 @@ public class RabbitMQHostedService : BackgroundService /// /// 将失败消息发送到死信队列 /// - private async Task SendToDeadLetterQueueAsync(string exchange, string dlqName, string dlqRoutingKey, byte[] body, CancellationToken cancellationToken) + private async Task SendToDeadLetterQueueAsync(string exchange, string dlqName, string dlqRoutingKey, string originalQueueName, byte[] body, Exception exception, CancellationToken cancellationToken) { try { @@ -111,7 +111,14 @@ public class RabbitMQHostedService : BackgroundService var properties = new RabbitMQ.Client.BasicProperties { - Persistent = true + Persistent = true, + Headers = new Dictionary + { + ["x-original-queue"] = originalQueueName, + ["x-error-type"] = exception.GetType().FullName, + ["x-error-message"] = exception.Message, + ["x-failed-at"] = DateTimeOffset.UtcNow.ToString("O") + } }; await channel.BasicPublishAsync(exchange, dlqRoutingKey, false, properties, body, cancellationToken); diff --git a/QYZH.InteractiveMagazine.WorkService/Jobs/MessageOutboxDispatchService.cs b/QYZH.InteractiveMagazine.WorkService/Jobs/MessageOutboxDispatchService.cs new file mode 100644 index 0000000..808bbdc --- /dev/null +++ b/QYZH.InteractiveMagazine.WorkService/Jobs/MessageOutboxDispatchService.cs @@ -0,0 +1,106 @@ +using QYZH.InteractiveMagazine.Infrastructure.RabbitMQ; +using QYZH.InteractiveMagazine.Models.Entity; +using QYZH.InteractiveMagazine.Models.Enum; +using QYZH.InteractiveMagazine.Models.Settings; +using SqlSugar; +using System.Text.Json; + +namespace QYZH.InteractiveMagazine.WorkService.Jobs; + +/// +/// 消息Outbox派发服务。 +/// +public class MessageOutboxDispatchService( + IServiceScopeFactory scopeFactory, + IRabbitMQService rabbitMQService, + IConfiguration configuration, + ILogger logger) : BackgroundService +{ + private const int BatchSize = 50; + private readonly RabbitMQRetrySettings retrySettings = configuration.GetSection("RabbitMQRetrySettings").Get() ?? new RabbitMQRetrySettings(); + + /// + /// 执行Outbox派发循环。 + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + while (!stoppingToken.IsCancellationRequested) + { + try + { + await DispatchPendingMessagesAsync(stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + logger.LogError(ex, "Outbox消息派发循环异常"); + } + + await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); + } + } + + private async Task DispatchPendingMessagesAsync(CancellationToken cancellationToken) + { + using var scope = scopeFactory.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var now = DateTime.Now; + var messages = await db.Queryable() + .Where(x => !x.IsDeleted + && (x.Status == (int)MessageOutboxStatusEnum.Pending || x.Status == (int)MessageOutboxStatusEnum.Failed) + && (x.NextRetryAt == null || x.NextRetryAt <= now)) + .OrderBy(x => x.CreatedAt) + .Take(BatchSize) + .ToListAsync(cancellationToken); + + foreach (var message in messages) + { + await DispatchMessageAsync(db, message, cancellationToken); + } + } + + private async Task DispatchMessageAsync(ISqlSugarClient db, MessageOutbox message, CancellationToken cancellationToken) + { + try + { + using var payload = JsonDocument.Parse(message.Payload); + var sent = await rabbitMQService.SendAsync(new RabbitMQSendParam + { + Exchange = message.Exchange, + Queue = message.Queue, + RoutingKey = message.RoutingKey, + Data = payload.RootElement.Clone() + }, cancellationToken); + + if (!sent) + { + throw new InvalidOperationException("RabbitMQ SendAsync returned false"); + } + + await db.Updateable() + .SetColumns(x => x.Status == (int)MessageOutboxStatusEnum.Sent) + .SetColumns(x => x.SentAt == DateTime.Now) + .SetColumns(x => x.UpdatedAt == DateTime.Now) + .Where(x => x.Id == message.Id && x.Status != (int)MessageOutboxStatusEnum.Sent) + .ExecuteCommandAsync(cancellationToken); + } + catch (Exception ex) + { + var retryCount = message.RetryCount + 1; + var abandoned = retryCount >= retrySettings.MaxRetryCount; + await db.Updateable() + .SetColumns(x => x.Status == (int)(abandoned ? MessageOutboxStatusEnum.Abandoned : MessageOutboxStatusEnum.Failed)) + .SetColumns(x => x.RetryCount == retryCount) + .SetColumns(x => x.NextRetryAt == (abandoned ? null : DateTime.Now.AddMilliseconds(retrySettings.RetryDelayMilliseconds))) + .SetColumns(x => x.LastError == ex.Message) + .SetColumns(x => x.UpdatedAt == DateTime.Now) + .Where(x => x.Id == message.Id) + .ExecuteCommandAsync(cancellationToken); + + logger.LogError(ex, "Outbox消息派发失败，MessageId: {MessageId}, RetryCount: {RetryCount}", message.Id, retryCount); + } + } +} diff --git a/QYZH.InteractiveMagazine.WorkService/Program.cs b/QYZH.InteractiveMagazine.WorkService/Program.cs index 162a35e..0908396 100644 --- a/QYZH.InteractiveMagazine.WorkService/Program.cs +++ b/QYZH.InteractiveMagazine.WorkService/Program.cs @@ -28,8 +28,8 @@ Log.Logger = new LoggerConfiguration() .CreateLogger(); builder.Services.AddSerilog(); -// 初始化雪花ID生成器 -YitIdHelper.SetIdGenerator(new IdGeneratorOptions { WorkerId = 2 }); +var snowflakeSettings = builder.Configuration.GetSection("SnowflakeSettings").Get() ?? new SnowflakeSettings { WorkerId = 2 }; +YitIdHelper.SetIdGenerator(new IdGeneratorOptions { WorkerId = snowflakeSettings.WorkerId }); // 初始化MySQL（SqlSugar） builder.Services.AddSqlSugar(new IocConfig @@ -81,6 +81,7 @@ builder.Services.AddScoped(); // 注册消费者后台服务 builder.Services.AddHostedService(); +builder.Services.AddHostedService(); // 从配置文件读取定时任务列表 var jobSettings = builder.Configuration.GetSection("HangfireJobs").Get(); diff --git a/QYZH.InteractiveMagazine.WorkService/appsettings.json b/QYZH.InteractiveMagazine.WorkService/appsettings.json index f0fe8f8..ddd4eac 100644 --- a/QYZH.InteractiveMagazine.WorkService/appsettings.json +++ b/QYZH.InteractiveMagazine.WorkService/appsettings.json @@ -2,6 +2,9 @@ "ConnectionStrings": { "DefaultConnection": "server=192.168.20.150;port=13306;database=InteractiveMagazine;user=user;password=n68792bu!y99r905;charset=utf8mb4;" }, + "SnowflakeSettings": { + "WorkerId": 2 + }, "RedisSettings": { "ConnectionString": "192.168.20.150:16379,defaultDatabase=5", "Sentinels": [], @@ -15,6 +18,13 @@ "VirtualHost": "InteractiveMagazine", "PrefetchCount": 1 }, + "RabbitMQRetrySettings": { + "MaxRetryCount": 3, + "RetryDelayMilliseconds": 30000 + }, + "HangfireStorageSettings": { + "StorageType": "Memory" + }, "Serilog": { "MinimumLevel": { "Default": "Information", @@ -52,7 +62,7 @@ "Name": "journal-task-ai-score-job", "JobType": "QYZH.InteractiveMagazine.WorkService.Jobs.JournalTaskAiScoreJob", "MethodName": "ExecuteAsync", - "Cron": "*/5 * * * *", + "Cron": "*/30 * * * *", "Enabled": true, "Description": "每30分钟扫描上次执行到本次执行之间的期刊答题记录并提交AI批改" }