word = "test" };
using (var connection = factory.CreateConnection())
{
while (Console.ReadLine() != null)
{
using (var channel = connection.CreateModel())
{
Dictionary<string, object> dic = new Dictionary<string, object>();
dic.Add("x-expires", 30000);
dic.Add("x-message-ttl", 12000);//队列上消息过期时间，应小于队列过期时间
dic.Add("x-dead-letter-exchange", "exchange-direct");//过期消息转向路由
dic.Add("x-dead-letter-routing-key", "routing-delay");//过期消息转向路由相匹配routingkey
//创建一个名叫"zzhello"的消息队列
channel.QueueDeclare(queue: "zzhello",
durable: true,
exclusive: false,
autoDelete: false,
arguments: dic);
var message = "Hello World!";
var body = Encoding.UTF8.GetBytes(message);
//向该消息队列发送消息message
channel.BasicPublish(exchange: "",
routingKey: "zzhello",
basicProperties: null,
body: body);
Console.WriteLine(" [x] Sent {0}", message);
}
}
}
Console.ReadKey();
消费者代码如下：
var factory = new ConnectionFactory() { HostName = "127.0.01", UserName = "test", Password = "test" };
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(exchange: "exchange-direct", type: "direct");
string name = channel.QueueDeclare().QueueName;
channel.QueueBind(queue: name, exchange: "exchange-direct", routingKey: "routing-delay");
//回调，当consumer收到消息后会执行该函数
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(ea.RoutingKey);
Console.WriteLine(" [x] Received {0}", message);
};
//Console.WriteLine("name:" + name);
//消费队列"hello"中的消息
channel.BasicConsume(queue: name,
autoAck: true,
consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
Console.ReadKey();
效果 ：
在等待了12秒后消费者等到了消息。
