何为Dead letter
max_length:
1)queue长度限制 【mongodg oplog】就是一个固定集合
2)queue中的message过期时间
3)basicreject basicnack等等
一、Dead letter exchange
有时候我们不希望message被drop掉,而是走到另外一个队列中,又或者是保存起来
二、Dead letter routing key
打到exchange中去的时候,确定routingkey的值
代码演示如下:
1、生产者
//基础配置 ConnectionFactory factory = new ConnectionFactory() { HostName = "10.123.44.12", UserName = "datamip", Password = "datamip" }; //第一步:创建connection using (var connection = factory.CreateConnection()) { //第二步:创建channel using (var channel = connection.CreateModel()) { channel.QueueDeclare("mytest", false, false, false, new Dictionary<string, object>() { {"x-max-length",10 }, { "x-dead-letter-exchange","mydead_exchange"}, {"x-dead-letter-routing-key","mydead_queue" } }); for (int i = 0; i < 15; i++) { channel.BasicPublish(string.Empty, "mytest", null, Encoding.UTF8.GetBytes(string.Format("你好{0}", i))); } } } Console.WriteLine("生产成功!"); Console.ReadKey();
2、消费者
//创建连接工厂 ConnectionFactory factory = new ConnectionFactory { UserName = "datamip",//用户名 Password = "datamip",//密码 HostName = "10.123.44.12"//rabbitmq ip }; //第一步:创建connection using (var connection = factory.CreateConnection()) { //第二步:创建channel using (var channel = connection.CreateModel()) { channel.ExchangeDeclare("mydead_exchange", ExchangeType.Direct, false, false, null); channel.QueueDeclare("mydead_queue",true,false,false,null); channel.QueueBind("mydead_queue", "mydead_exchange", "mydead_queue", null); EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received +=(sender,e)=> { var msg = Encoding.UTF8.GetString(e.Body.ToArray()); Console.WriteLine(msg); }; channel.BasicConsume("mydead_queue",true, consumer); Console.ReadKey(); } }
三、Maximum priority
优先级加载队列上,就是一个优先级队列 【堆,大根堆,小根堆】
1、给队列贴上一个优先级的标签 x-max-priority=10
2、给message打上具体的优先级的value value<=10
代码演示:
1、生产者
//基础配置 ConnectionFactory factory = new ConnectionFactory() { HostName = "10.123.44.12", UserName = "datamip", Password = "datamip" }; //第一步:创建connection using (var connection = factory.CreateConnection()) { //第二步:创建channel using (var channel = connection.CreateModel()) { channel.QueueDeclare("mytest", false, false, false, new Dictionary<string, object>() { {"x-max-priority",10 } }); var properties = channel.CreateBasicProperties(); for (int i = 0; i < 10; i++) { properties.Priority = (byte)i; channel.BasicPublish(string.Empty, "mytest", properties, Encoding.UTF8.GetBytes(string.Format("你好{0}", i))); } } } Console.WriteLine("生产成功!"); Console.ReadKey();
2、消费者
//创建连接工厂 ConnectionFactory factory = new ConnectionFactory { UserName = "datamip",//用户名 Password = "datamip",//密码 HostName = "10.123.44.12"//rabbitmq ip }; //第一步:创建connection using (var connection = factory.CreateConnection()) { //第二步:创建channel using (var channel = connection.CreateModel()) { EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received +=(sender,e)=> { var msg = Encoding.UTF8.GetString(e.Body.ToArray()); Console.WriteLine(msg); }; channel.BasicConsume("mytest", true, consumer); Console.ReadKey(); } }