.NET中使用RabbitMQ延时队列和死信队列 - qidao123.com技术社区-IT企服评测·应用市场

var args = new Dictionary<string,object>();
args.Add("x-message-ttl",60000); //单位为毫秒
model.QueueDeclare("myqueue",false,false,false,args);

复制代码

IBasicProperties props = model.CreateBasicProperties();
props.ContentType = "text/plain";
props.DeliveryMode = 2;
props.Expiration = "60000"
model.BasicPublic(exchangeName,routingKey,props,messageBodyBytes);

复制代码

public class OrderDto
{
/// <summary>
/// 订单名称
/// </summary>
public string Name { get; set; }
/// <summary>
/// 订单状态
/// 0 未支付
/// 1 已支付
/// 2 超时
/// </summary>
public int Status { get; set; }
}

复制代码

[ApiController]
[Route("api/orders")]
public class OrdersController : ControllerBase
{
private readonly IOrderService _orderService;
public OrdersController(IOrderService orderService)
{
_orderService = orderService;
}
[HttpPost]
public IActionResult CreateOrder([FromBody] OrderDto orderDto)
{
// 处理订单逻辑
_orderService.ProcessOrder(orderDto);
return Ok();
}
}

复制代码

public interface IOrderService
{
void ProcessOrder(OrderDto orderDto);
}
public class OrderService : IOrderService
{
private readonly RabbitMQConnectionFactory _connectionFactory;
public OrderService(RabbitMQConnectionFactory connectionFactory)
{
_connectionFactory = connectionFactory;
}
public void ProcessOrder(OrderDto orderDto)
{
using (var channel = _connectionFactory.CreateChannel())
{
var properties = channel.CreateBasicProperties();
properties.Headers = new Dictionary<string, object>
{
{ "x-delay", 1000 * 20 } // 设置20秒延时
};
var message = JsonConvert.SerializeObject(orderDto);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish("delayed_exchange", "routing_key", properties, body);
}
}
}

复制代码

public class ProcessPay : IHostedService
{
private readonly ConnectionFactory _factory;
private IConnection _connection;
private IModel _channel;
public ProcessPay()
{
_factory = new ConnectionFactory()
{
HostName = "ip",
Port = 5672,
UserName = "用户名",
Password = "密码"
};
}
public Task StartAsync(CancellationToken cancellationToken)
{
Console.WriteLine(" Press [enter] to exit.");
_connection = _factory.CreateConnection();
_channel = _connection.CreateModel();
_channel.ExchangeDeclare("delayed_exchange", ExchangeType.Direct, true, false, null);
//关键代码，绑定死信交换机
var arguments = new Dictionary<string, object>
{
{ "x-dead-letter-exchange", "dead_letter_exchange" },
{ "x-dead-letter-routing-key", "dead_letter_routing_key" }
};
_channel.QueueDeclare("delayed_queue", true, false, false, arguments);
_channel.QueueBind("delayed_queue", "delayed_exchange", "routing_key");
var consumer = new EventingBasicConsumer(_channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
// 处理支付逻辑
var orderDto = JsonConvert.DeserializeObject<OrderDto>(message);
Console.WriteLine($"订单信息:{orderDto.Name}");
Console.WriteLine("请输入价格（模拟支付）：");
// 超时未支付
string? many = "";
// 支付处理
Console.WriteLine("请输入:");
// 超时未支付进行处理
Task.Factory.StartNew(() =>
{
many = Console.ReadLine();
Console.WriteLine($"many:{many}");
}).Wait(20 * 1000);
if (string.Equals(many, "100"))
{
orderDto.Status = 1;
Console.WriteLine("支付完成");
_channel.BasicAck(ea.DeliveryTag, true);
}
else
{
//重试几次依然失败
Console.WriteLine("等待一定时间内失效超时未支付的订单");
_channel.BasicNack(ea.DeliveryTag, false, false);
}
};
_channel.BasicConsume("delayed_queue", false, consumer);
return Task.CompletedTask;
}
public Task StopAsync(CancellationToken cancellationToken)
{
_channel?.Close();
_connection?.Close();
_channel?.Dispose();
_connection?.Dispose();
return Task.CompletedTask;
}
}

复制代码

static void Main(string[] args)
{
var host = new HostBuilder()
.ConfigureServices((hostContext, services) =>
{
services.AddSingleton<IHostedService,ProcessPay>();
})
.Build();
host.Run();
}

复制代码

using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare("dead_letter_exchange", ExchangeType.Direct, true, false, null);
channel.QueueDeclare("dead_letter_queue", true, false, false, null);
channel.QueueBind("dead_letter_queue", "dead_letter_exchange", "dead_letter_routing_key");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
// 处理超时未支付的订单逻辑
var orderDto = JsonConvert.DeserializeObject<OrderDto>(message);
orderDto.Status = 2;
Console.WriteLine($"订单信息:{orderDto.Name},{orderDto.Status}");
Console.WriteLine("超时未支付");
channel.BasicAck(ea.DeliveryTag, true);
};
channel.BasicConsume("dead_letter_queue", false, consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}

复制代码