refactor: 拆分自动铺码服务为独立PrintWorker项目
1. 将原WorkService中的AutoDotCodeConsumer、RabbitMQHostedService、IQueueConsumer迁移至新的PrintWorker项目 2. 移除WorkService中冗余的PrintTool依赖文件复制配置 3. 将PrintToolV2.7相关资源移动到PrintWorker项目中统一管理 4. 从WorkService中移除AutoDotCodeConsumer的服务注册 5. 新增PrintWorker项目的完整宿主程序与配置文件
This commit is contained in:
@ -0,0 +1,338 @@
|
||||
using QYZH.InteractiveMagazine.Common.Extensions;
|
||||
using QYZH.InteractiveMagazine.Infrastructure.OSS;
|
||||
using QYZH.InteractiveMagazine.Models.Dto.Journal;
|
||||
using QYZH.InteractiveMagazine.Models.Entity;
|
||||
using QYZH.InteractiveMagazine.Models.Enum;
|
||||
using SqlSugar;
|
||||
using System.Diagnostics;
|
||||
using System.Net;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace QYZH.InteractiveMagazine.PrintWorker.Consumers;
|
||||
|
||||
/// <summary>
|
||||
/// 自动铺码消费者。
|
||||
/// </summary>
|
||||
public class AutoDotCodeConsumer(
|
||||
IConfiguration configuration,
|
||||
IServiceScopeFactory scopeFactory,
|
||||
IHostEnvironment hostEnvironment,
|
||||
ILogger<AutoDotCodeConsumer> logger,
|
||||
IHttpClientFactory httpClientFactory,
|
||||
OssService ossService) : IQueueConsumer
|
||||
{
|
||||
/// <summary>
|
||||
/// 交换机名称。
|
||||
/// </summary>
|
||||
public string Exchange => "ex.journal";
|
||||
|
||||
/// <summary>
|
||||
/// 队列名称。
|
||||
/// </summary>
|
||||
public string QueueName => "mq.journal.dotcode.auto";
|
||||
|
||||
/// <summary>
|
||||
/// 路由键。
|
||||
/// </summary>
|
||||
public string RoutingKey => "rk.journal.dotcode.auto";
|
||||
|
||||
/// <summary>
|
||||
/// 处理自动铺码消息。
|
||||
/// </summary>
|
||||
public async Task HandleAsync(byte[] body, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var message = Encoding.UTF8.GetString(body);
|
||||
logger.LogInformation("收到自动铺码消息: {Message}", message);
|
||||
|
||||
using var scope = scopeFactory.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
|
||||
|
||||
try
|
||||
{
|
||||
var request = JsonSerializer.Deserialize<JournalPagePrintDto>(message, new JsonSerializerOptions
|
||||
{
|
||||
PropertyNameCaseInsensitive = true
|
||||
});
|
||||
if (request == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var workDirectory = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "JournalPagePdf");
|
||||
var uploadDirectory = Path.Combine(workDirectory, "upload");
|
||||
var downloadDirectory = Path.Combine(workDirectory, "download");
|
||||
Directory.CreateDirectory(uploadDirectory);
|
||||
Directory.CreateDirectory(downloadDirectory);
|
||||
|
||||
var uploadFileName = $"upload_{request.JournalId}_{DateTime.Now:yyyyMMddHHmmssffffff}{Random.Shared.Next(1000, 9999)}.pdf";
|
||||
var uploadFilePath = Path.Combine(uploadDirectory, uploadFileName);
|
||||
|
||||
var journalPdfKey = request.JournalPdfUrl?.RemoveDomain();
|
||||
if (string.IsNullOrWhiteSpace(journalPdfKey))
|
||||
{
|
||||
logger.LogError("书籍上传 PDF 文件地址为空，JournalId: {JournalId}", request.JournalId);
|
||||
return;
|
||||
}
|
||||
|
||||
await using var pdfStream = ossService.GetObjectStream(journalPdfKey);
|
||||
if (pdfStream == null)
|
||||
{
|
||||
logger.LogError("获取书籍上传 PDF 文件失败，OSS Key: {OssKey}", journalPdfKey);
|
||||
return;
|
||||
}
|
||||
|
||||
await using (var fs = new FileStream(uploadFilePath, FileMode.CreateNew, FileAccess.Write))
|
||||
{
|
||||
await pdfStream.CopyToAsync(fs, cancellationToken);
|
||||
logger.LogInformation("书籍 PDF 已下载到本地，长度：{Length}", fs.Length);
|
||||
}
|
||||
|
||||
var dotId = configuration.GetValue("PrintConfig:DotId", 765837655859269);
|
||||
var dotFile = await dbContext.Queryable<DotFile>().FirstAsync(x => x.Id == dotId, cancellationToken);
|
||||
if (dotFile == null)
|
||||
{
|
||||
logger.LogError("配置打印数据错误，点阵文件不存在，DotId: {DotId}", dotId);
|
||||
return;
|
||||
}
|
||||
|
||||
var printToolDirectory = Path.Combine(hostEnvironment.ContentRootPath, "PrintToolV2.7");
|
||||
var exePath = Path.Combine(printToolDirectory, "PrintTool.exe");
|
||||
if (!File.Exists(exePath))
|
||||
{
|
||||
logger.LogError("PrintTool.exe 不存在，路径：{ExePath}", exePath);
|
||||
await ExecuteUpdateJournalStatusAsync(BuildFailResponse(request), cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
var xmlPath = Path.Combine(printToolDirectory, dotFile.FileName);
|
||||
if (!File.Exists(xmlPath))
|
||||
{
|
||||
logger.LogError("铺码授权文件不存在，路径：{XmlPath}", xmlPath);
|
||||
await ExecuteUpdateJournalStatusAsync(BuildFailResponse(request), cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
var downloadFileName = $"download_{request.JournalId}_{DateTime.Now:yyyyMMddHHmmssffffff}{Random.Shared.Next(1000, 9999)}_dot.pdf";
|
||||
var downloadFilePath = Path.Combine(downloadDirectory, downloadFileName);
|
||||
var dPrint = configuration.GetValue("PrintConfig:DPrint", 0);
|
||||
var pageNumMax = request.PageNum.Max(x => x);
|
||||
var dotFileDetailList = await dbContext.Queryable<DotFileDetail>()
|
||||
.Where(x => x.DotId == dotId && !x.IsUse)
|
||||
.OrderBy(x => x.Id)
|
||||
.Take(pageNumMax)
|
||||
.ToListAsync(cancellationToken);
|
||||
|
||||
if (dotFileDetailList.Count < pageNumMax)
|
||||
{
|
||||
logger.LogError("点阵页码余量不足，DotId: {DotId}, Need: {Need}, Available: {Available}", dotId, pageNumMax, dotFileDetailList.Count);
|
||||
await ExecuteUpdateJournalStatusAsync(BuildFailResponse(request), cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
var dotFileDetailPageName = dotFileDetailList.Select(x => x.PageName).OrderBy(x => x).ToArray();
|
||||
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);
|
||||
|
||||
if (printResult.Timeout || printResult.ExitCode != 0)
|
||||
{
|
||||
logger.LogError("执行 PrintTool.exe 失败，退出码：{ExitCode}，错误信息：{ErrorMessage}", printResult.ExitCode, printResult.Error);
|
||||
await ExecuteUpdateJournalStatusAsync(BuildFailResponse(request, dotFileDetailPageName), cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
var statusModel = new JournalPagePrintDto
|
||||
{
|
||||
JournalId = request.JournalId,
|
||||
PageNo = dotFileDetailPageName
|
||||
};
|
||||
|
||||
if (!ValidatePrintOutput(printResult.Output, dotFileDetailPageName, dPrint))
|
||||
{
|
||||
statusModel.Status = JournalStatusEnum.CodeFail;
|
||||
await ExecuteUpdateJournalStatusAsync(statusModel, cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
var tempOssDownloadKey = "journal/download/" + downloadFileName;
|
||||
const int bufferSize = 1 * 1024 * 1024;
|
||||
await using var downloadFs = new FileStream(downloadFilePath, FileMode.Open, FileAccess.Read, FileShare.Read, bufferSize, FileOptions.SequentialScan | FileOptions.Asynchronous);
|
||||
var downloadBookPagePdfName = ossService.PutObject(tempOssDownloadKey, downloadFs);
|
||||
var ossDomain = configuration.GetSection("AliyunOSSConfigs:Domain").Get<string>() ?? string.Empty;
|
||||
|
||||
statusModel.DownloadJournalPagePdfName = ossDomain + downloadBookPagePdfName;
|
||||
await downloadFs.DisposeAsync();
|
||||
|
||||
TryDeleteFile(uploadFilePath);
|
||||
TryDeleteFile(downloadFilePath);
|
||||
|
||||
statusModel.Status = JournalStatusEnum.CodeSuccess;
|
||||
var callbackResponse = await ExecuteUpdateJournalStatusAsync(statusModel, cancellationToken);
|
||||
if (callbackResponse.IsSuccess)
|
||||
{
|
||||
await dbContext.Ado.UseTranAsync(async () =>
|
||||
{
|
||||
foreach (var dotFileDetail in dotFileDetailList)
|
||||
{
|
||||
dotFileDetail.IsUse = true;
|
||||
dotFileDetail.UpdatedAt = DateTime.Now;
|
||||
}
|
||||
|
||||
await dbContext.Updateable(dotFileDetailList).ExecuteCommandAsync();
|
||||
await dbContext.Updateable<DotFile>()
|
||||
.SetColumns(x => x.TotalUse == x.TotalUse + dotFileDetailList.Count)
|
||||
.SetColumns(x => x.UpdatedAt == DateTime.Now)
|
||||
.Where(x => x.Id == dotId)
|
||||
.ExecuteCommandAsync();
|
||||
});
|
||||
}
|
||||
else
|
||||
{
|
||||
logger.LogError("回调接口修改书籍状态失败，JournalId: {JournalId}，接口返回消息：{Message}", statusModel.JournalId, callbackResponse.Message);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "自动铺码处理失败");
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 处理消费异常。
|
||||
/// </summary>
|
||||
public Task OnErrorAsync(byte[] message, Exception exception)
|
||||
{
|
||||
var body = Encoding.UTF8.GetString(message);
|
||||
logger.LogError(exception, "处理自动铺码消息失败: {Message}", body);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private async Task<PrintToolResult> ExecutePrintToolAsync(string exePath, string workingDirectory, string arguments, CancellationToken cancellationToken)
|
||||
{
|
||||
var printToolTimeoutSeconds = configuration.GetValue("PrintConfig:TimeoutSeconds", 300);
|
||||
var printToolTimeout = TimeSpan.FromSeconds(printToolTimeoutSeconds);
|
||||
var cmd = $"\"{exePath}\" {arguments}";
|
||||
logger.LogInformation("执行 PrintTool.exe 的命令: {Command}，超时时间：{TimeoutSeconds} 秒", cmd, printToolTimeoutSeconds);
|
||||
|
||||
using var process = new Process();
|
||||
process.StartInfo = new ProcessStartInfo
|
||||
{
|
||||
WorkingDirectory = workingDirectory,
|
||||
FileName = exePath,
|
||||
Arguments = arguments,
|
||||
UseShellExecute = false,
|
||||
RedirectStandardOutput = true,
|
||||
RedirectStandardError = true,
|
||||
CreateNoWindow = true
|
||||
};
|
||||
|
||||
process.Start();
|
||||
var outputTask = process.StandardOutput.ReadToEndAsync(cancellationToken);
|
||||
var errorTask = process.StandardError.ReadToEndAsync(cancellationToken);
|
||||
var timeout = false;
|
||||
|
||||
try
|
||||
{
|
||||
await process.WaitForExitAsync(cancellationToken).WaitAsync(printToolTimeout, cancellationToken);
|
||||
}
|
||||
catch (TimeoutException)
|
||||
{
|
||||
timeout = true;
|
||||
logger.LogError("执行 PrintTool.exe 超时，准备终止进程树，超时时间：{TimeoutSeconds} 秒，命令：{Command}", printToolTimeoutSeconds, cmd);
|
||||
process.Kill(entireProcessTree: true);
|
||||
await process.WaitForExitAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
var output = await outputTask;
|
||||
var error = await errorTask;
|
||||
var exitCode = process.ExitCode;
|
||||
|
||||
logger.LogInformation("执行 PrintTool.exe 后，退出码：{ExitCode}，stdout：{Output}", exitCode, output);
|
||||
if (!string.IsNullOrWhiteSpace(error))
|
||||
{
|
||||
logger.LogWarning("执行 PrintTool.exe 后，stderr：{ErrorMessage}", error);
|
||||
}
|
||||
|
||||
return new PrintToolResult(output, error, exitCode, timeout);
|
||||
}
|
||||
|
||||
private bool ValidatePrintOutput(string output, string[] dotFileDetailPageName, int dPrint)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(output))
|
||||
{
|
||||
logger.LogInformation("执行铺码程序没有任何输出");
|
||||
return false;
|
||||
}
|
||||
|
||||
if (dPrint == 0)
|
||||
{
|
||||
var pageNoList = JsonSerializer.Deserialize<string[]>(output) ?? [];
|
||||
return pageNoList.OrderBy(x => x).SequenceEqual(dotFileDetailPageName.OrderBy(x => x), StringComparer.Ordinal);
|
||||
}
|
||||
|
||||
return dotFileDetailPageName.AsParallel().Any(ip => output.Contains(ip, StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
private async Task<CallbackUpdateJournalStatusResponse> ExecuteUpdateJournalStatusAsync(JournalPagePrintDto request, CancellationToken cancellationToken)
|
||||
{
|
||||
var client = httpClientFactory.CreateClient();
|
||||
var callbackUrl = configuration.GetValue<string>("PrintConfig:CallBackApiUrl");
|
||||
var callbackRequest = new HttpRequestMessage(HttpMethod.Post, callbackUrl)
|
||||
{
|
||||
Content = new StringContent(JsonSerializer.Serialize(request), Encoding.UTF8, "application/json")
|
||||
};
|
||||
|
||||
var response = await client.SendAsync(callbackRequest, cancellationToken);
|
||||
if (response.StatusCode == HttpStatusCode.OK)
|
||||
{
|
||||
var responseContent = await response.Content.ReadAsStringAsync(cancellationToken);
|
||||
logger.LogInformation("回调接口返回内容：{ResponseContent}", responseContent);
|
||||
return JsonSerializer.Deserialize<CallbackUpdateJournalStatusResponse>(responseContent, new JsonSerializerOptions
|
||||
{
|
||||
PropertyNameCaseInsensitive = true
|
||||
}) ?? new CallbackUpdateJournalStatusResponse();
|
||||
}
|
||||
|
||||
return new CallbackUpdateJournalStatusResponse();
|
||||
}
|
||||
|
||||
private static JournalPagePrintDto BuildFailResponse(JournalPagePrintDto request, string[]? pageNo = null)
|
||||
{
|
||||
return new JournalPagePrintDto
|
||||
{
|
||||
JournalId = request.JournalId,
|
||||
PageNo = pageNo ?? [],
|
||||
Status = JournalStatusEnum.CodeFail
|
||||
};
|
||||
}
|
||||
|
||||
private void TryDeleteFile(string filePath)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (File.Exists(filePath))
|
||||
{
|
||||
File.Delete(filePath);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogWarning(ex, "删除临时文件失败，路径：{FilePath}", filePath);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed record PrintToolResult(string Output, string Error, int ExitCode, bool Timeout);
|
||||
|
||||
internal class CallbackUpdateJournalStatusResponse
|
||||
{
|
||||
public string? Message { get; set; }
|
||||
|
||||
public string? Code { get; set; }
|
||||
|
||||
public bool Result { get; set; }
|
||||
|
||||
public bool IsSuccess { get; set; }
|
||||
}
|
||||
@ -0,0 +1,32 @@
|
||||
namespace QYZH.InteractiveMagazine.PrintWorker.Consumers;
|
||||
|
||||
/// <summary>
|
||||
/// 队列消费者接口。
|
||||
/// </summary>
|
||||
public interface IQueueConsumer
|
||||
{
|
||||
/// <summary>
|
||||
/// 交换机名称。
|
||||
/// </summary>
|
||||
string Exchange { get; }
|
||||
|
||||
/// <summary>
|
||||
/// 队列名称。
|
||||
/// </summary>
|
||||
string QueueName { get; }
|
||||
|
||||
/// <summary>
|
||||
/// 路由键。
|
||||
/// </summary>
|
||||
string RoutingKey { get; }
|
||||
|
||||
/// <summary>
|
||||
/// 处理消息。
|
||||
/// </summary>
|
||||
Task HandleAsync(byte[] message, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// 处理消费异常。
|
||||
/// </summary>
|
||||
Task OnErrorAsync(byte[] message, Exception exception);
|
||||
}
|
||||
@ -0,0 +1,105 @@
|
||||
using QYZH.InteractiveMagazine.Infrastructure.RabbitMQ;
|
||||
|
||||
namespace QYZH.InteractiveMagazine.PrintWorker.Consumers;
|
||||
|
||||
/// <summary>
|
||||
/// RabbitMQ 消费者后台服务。
|
||||
/// </summary>
|
||||
public class RabbitMQHostedService(IServiceProvider serviceProvider, ILogger<RabbitMQHostedService> logger) : BackgroundService
|
||||
{
|
||||
/// <summary>
|
||||
/// 启动所有已注册队列消费者。
|
||||
/// </summary>
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
using var scope = serviceProvider.CreateScope();
|
||||
var consumers = scope.ServiceProvider.GetServices<IQueueConsumer>().ToList();
|
||||
|
||||
if (consumers.Count == 0)
|
||||
{
|
||||
logger.LogWarning("未注册任何队列消费者");
|
||||
return;
|
||||
}
|
||||
|
||||
var tasks = consumers.Select(c => StartConsumerAsync(c, stoppingToken)).ToList();
|
||||
await Task.WhenAll(tasks);
|
||||
}
|
||||
|
||||
private async Task StartConsumerAsync(IQueueConsumer queueConsumer, CancellationToken stoppingToken)
|
||||
{
|
||||
var queueName = queueConsumer.QueueName;
|
||||
var exchange = queueConsumer.Exchange;
|
||||
var routingKey = queueConsumer.RoutingKey;
|
||||
logger.LogInformation("正在启动消费者 {QueueName}, Exchange: {Exchange}, RoutingKey: {RoutingKey}", queueName, exchange, routingKey);
|
||||
|
||||
try
|
||||
{
|
||||
var rabbitMQService = serviceProvider.GetRequiredService<IRabbitMQService>();
|
||||
await rabbitMQService.ReceiveAsync(exchange, queueName, routingKey, async (channel, ea) =>
|
||||
{
|
||||
var dlqName = $"{queueName}.dlq";
|
||||
var dlqRoutingKey = $"{routingKey}.dlq";
|
||||
await channel.QueueDeclareAsync(queue: dlqName, durable: true, exclusive: false, autoDelete: false, arguments: null);
|
||||
await channel.QueueBindAsync(queue: dlqName, exchange: exchange, routingKey: dlqRoutingKey, arguments: null);
|
||||
|
||||
using var messageScope = serviceProvider.CreateScope();
|
||||
var consumer = messageScope.ServiceProvider
|
||||
.GetServices<IQueueConsumer>()
|
||||
.First(c => c.QueueName == queueName);
|
||||
|
||||
try
|
||||
{
|
||||
await consumer.HandleAsync(ea.Body.ToArray(), stoppingToken);
|
||||
await channel.BasicAckAsync(ea.DeliveryTag, false, stoppingToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "消费者 {QueueName} 处理消息异常", queueName);
|
||||
var dlqSent = await SendToDeadLetterQueueAsync(exchange, dlqName, dlqRoutingKey, ea.Body.ToArray(), stoppingToken);
|
||||
await channel.BasicNackAsync(ea.DeliveryTag, false, !dlqSent, stoppingToken);
|
||||
|
||||
try
|
||||
{
|
||||
await consumer.OnErrorAsync(ea.Body.ToArray(), ex);
|
||||
}
|
||||
catch (Exception errorEx)
|
||||
{
|
||||
logger.LogError(errorEx, "消费者 {QueueName} OnErrorAsync 执行异常", queueName);
|
||||
}
|
||||
}
|
||||
}, stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
logger.LogInformation("消费者 {QueueName} 已停止", queueName);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "消费者 {QueueName} 启动失败", queueName);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<bool> SendToDeadLetterQueueAsync(string exchange, string dlqName, string dlqRoutingKey, byte[] body, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
var rabbitMQConnection = serviceProvider.GetRequiredService<IRabbitMQConnection>();
|
||||
using var channel = await rabbitMQConnection.CreateChannel();
|
||||
await channel.QueueDeclareAsync(queue: dlqName, durable: true, exclusive: false, autoDelete: false, arguments: null);
|
||||
await channel.QueueBindAsync(queue: dlqName, exchange: exchange, routingKey: dlqRoutingKey, arguments: null);
|
||||
|
||||
var properties = new RabbitMQ.Client.BasicProperties
|
||||
{
|
||||
Persistent = true
|
||||
};
|
||||
await channel.BasicPublishAsync(exchange, dlqRoutingKey, false, properties, body, cancellationToken);
|
||||
logger.LogInformation("消息已发送到死信队列: {DlqName}", dlqName);
|
||||
return true;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "发送消息到死信队列 {DlqName} 失败", dlqName);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user