我目前正在玩Rabbit-Mq,并且正在尝试实现“死信”队列,即失败消息的队列。我一直在阅读Rabbit文档:https : //www.rabbitmq.com/dlx.html。
并提出了以下示例:
internal class Program
{
private const string WorkerExchange = "work.exchange";
private const string RetryExchange = "retry.exchange";
public const string WorkerQueue = "work.queue";
private const string RetryQueue = "retry.queue";
static void Main(string[] args)
{
var factory = new ConnectionFactory { HostName = "localhost" };
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(WorkerExchange, "direct");
channel.QueueDeclare
(
WorkerQueue, true, false, false,
new Dictionary<string, object>
{
{"x-dead-letter-exchange", RetryExchange},
// I have tried with and without this next key
{"x-dead-letter-routing-key", RetryQueue}
}
);
channel.QueueBind(WorkerQueue, WorkerExchange, string.Empty, null);
channel.ExchangeDeclare(RetryExchange, "direct");
channel.QueueDeclare
(
RetryQueue, true, false, false,
new Dictionary<string, object> {
{ "x-dead-letter-exchange", WorkerExchange },
{ "x-message-ttl", 30000 },
}
);
channel.QueueBind(RetryQueue, RetryExchange, string.Empty, null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] Received {0}", message);
Thread.Sleep(1000);
Console.WriteLine("Rejected message");
// also tried channel.BasicNack(ea.DeliveryTag, false, false);
channel.BasicReject(ea.DeliveryTag, false);
};
channel.BasicConsume(WorkerQueue, false, consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}
}
Run Code Online (Sandbox Code Playgroud)
我感觉好像丢失了一些小细节,但似乎找不到它们。
提前致谢
您应将Dead-letter-exchange定义为fanout。
我们去了: channel.ExchangeDeclare(RetryExchange, "fanout");
如果将死信交换设置为DIRECT,则必须指定死信路由密钥。如果您只是想让所有被NACKed的消息都进入一个死信箱中,以便以后进行调查(就像我一样),则应该将死信交换设置为FANOUT。
事实证明,如果死信交换是direct交换,那么队列参数需要一个x-dead-letter-routing-key. 上面(在问题中)我在字典中使用这个键来尝试路由我的消息,但我没有做的是向我的绑定添加路由,这里是代码的更新版本:
internal class Program
{
private const string WorkerExchange = "work.exchange";
private const string RetryExchange = "retry.exchange";
public const string WorkerQueue = "work.queue";
private const string RetryQueue = "retry.queue";
static void Main(string[] args)
{
var factory = new ConnectionFactory { HostName = "localhost" };
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(WorkerExchange, "direct");
channel.QueueDeclare
(
WorkerQueue, true, false, false,
new Dictionary<string, object>
{
{"x-dead-letter-exchange", RetryExchange},
{"x-dead-letter-routing-key", RetryQueue}
}
);
channel.QueueBind(WorkerQueue, WorkerExchange, WorkerQueue, null);
channel.ExchangeDeclare(RetryExchange, "direct");
channel.QueueDeclare
(
RetryQueue, true, false, false,
new Dictionary<string, object>
{
{"x-dead-letter-exchange", WorkerExchange},
{"x-dead-letter-routing-key", WorkerQueue},
{"x-message-ttl", 30000},
}
);
channel.QueueBind(RetryQueue, RetryExchange, RetryQueue, null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] Received {0}", message);
Thread.Sleep(1000);
Console.WriteLine("Rejected message");
channel.BasicNack(ea.DeliveryTag, false, false);
};
channel.BasicConsume(WorkerQueue, false, consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
}
Run Code Online (Sandbox Code Playgroud)
不同之处在于调用channel.QueueBind(WorkerQueue, WorkerExchange, WorkerQueue, null);now 提供的路由键与队列名称相同,因此当消息“死信”时,它会通过此键路由到交换