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

refactor: 重构RabbitMQ消费者体系，新增自动铺码功能

Browse Source 
1. 新增AutoDotCodeConsumer自动铺码消费者，实现统一的交换机路由绑定
2. 重构RabbitMQ接收方法，支持交换机、路由键配置，添加死信队列处理
3. 重构现有JournalTaskReceiveConsumer，使用标准交换机路由配置
4. 更新JournalPageService，将打印逻辑替换为自动铺码消息发送
5. 调整枚举类型，重构任务类型命名
6. 更新实体和DTO，新增Prompt相关字段
7. 添加PrintToolV2.7配套工具和文档
8. 清理冗余的PrintJournalPageAsync接口
This commit is contained in:
glz
2026-06-22 16:56:05 +08:00
parent 6ece22d23d
commit d9933e4537
21 changed files with 252 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
QYZH.InteractiveMagazine.WorkService/Consumers/RabbitMQHostedService.cs
View File

@ -29,20 +29,29 @@ public class RabbitMQHostedService : BackgroundService
_logger.LogInformation("发现 {Count} 个队列消费者，开始启动...", consumers.Count);
var tasks = consumers.Select(c => StartConsumerAsync(c.QueueName, stoppingToken)).ToList();
var tasks = consumers.Select(c => StartConsumerAsync(c, stoppingToken)).ToList();
await Task.WhenAll(tasks);
}
private async Task StartConsumerAsync(string queueName, CancellationToken stoppingToken)
private async Task StartConsumerAsync(IQueueConsumer queueConsumer, CancellationToken stoppingToken)
{
_logger.LogInformation("正在启动消费者: {QueueName}", queueName);
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(queueName, async (channel, ea) =>
await rabbitMQService.ReceiveAsync(exchange, queueName, routingKey, async (channel, ea) =>
{
// 声明死信队列及绑定（幂等操作，确保DLQ存在）
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);
// 每条消息创建独立 scope，确保消费者内可注入 Scoped 服务
using var messageScope = _serviceProvider.CreateScope();
var consumer = messageScope.ServiceProvider
@ -57,7 +66,15 @@ public class RabbitMQHostedService : BackgroundService
catch (Exception ex)
{
_logger.LogError(ex, "消费者 {QueueName} 处理消息异常", queueName);
await channel.BasicNackAsync(ea.DeliveryTag, false, true, stoppingToken);
// 发送到死信队列，成功则从主队列移除，失败则重新入队
var dlqSent = await SendToDeadLetterQueueAsync(exchange, dlqName, dlqRoutingKey, ea.Body.ToArray(), stoppingToken);
await channel.BasicNackAsync(ea.DeliveryTag, false, !dlqSent, stoppingToken);
if (dlqSent)
{
_logger.LogInformation("消息已转入死信队列 {DlqName}", dlqName);
}
try
{
@ -79,4 +96,32 @@ public class RabbitMQHostedService : BackgroundService
_logger.LogError(ex, "消费者 {QueueName} 启动失败", queueName);
}
}
/// <summary>
/// 将失败消息发送到死信队列
/// </summary>
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;
}
}
}