?private?static?string?_exchangeNormal?=?"Exchange.Normal";??//定义一个用于接收?正常?消息的交换机
????????private?static?string?_exchangeRetry?=?"Exchange.Retry";????//定义一个用于接收?重试?消息的交换机
????????private?static?string?_exchangeFail?=?"Exchange.Fail";??????//定义一个用于接收?失败?消息的交换机
????????private?static?string?_queueNormal?=?"Queue.Noraml";????????//定义一个用于接收?正常?消息的队列
????????private?static?string?_queueRetry?=?"Queue.Retry";??????????//定义一个用于接收?重试?消息的队列
????????private?static?string?_queueFail?=?"Queue.Fail";????????????//定义一个用于接收?失败?消息的队列

????????public?static?void?Test()
????????{
????????????var?connection?=?RabbitMQHelper.GetConnection();
????????????var?channel?=?connection.CreateModel();

????????????//声明交换机
????????????channel.ExchangeDeclare(_exchangeNormal,?"topic",?true,?false,?null);
????????????channel.ExchangeDeclare(_exchangeRetry,?"topic",?true,?false,?null);
????????????channel.ExchangeDeclare(_exchangeFail,?"topic",?true,?false,?null);

????????????//定义队列参数
????????????var?queueNormalArgs?=?new?Dictionary<string,?object>();
????????????{
????????????????queueNormalArgs.Add("x-dead-letter-exchange",?_exchangeFail);???//指定死信交换机，用于将?Normal?队列中失败的消息投递给?Fail?交换机
????????????}
????????????var?queueRetryArgs?=?new?Dictionary<string,?object>();
????????????{
????????????????queueRetryArgs.Add("x-dead-letter-exchange",?_exchangeNormal);??//指定死信交换机，用于将?Retry?队列中超时的消息投递给?Normal?交换机
????????????????queueRetryArgs.Add("x-message-ttl",?6000);??????????????????????//定义 queueRetry 的消息最大停留时间?(原理是：等消息超时后由 broker 自动投递给当前绑定的死信交换机)?????????????????????????????????????????????????????????????????????????????//定义最大停留时间为防止一些?待重新投递?的消息、没有定义重试时间而导致内存溢出
????????????}
????????????var?queueFailArgs?=?new?Dictionary<string,?object>();
????????????{
????????????}

????????????//声明队列
????????????channel.QueueDeclare(queue:?_queueNormal,?durable:?true,?exclusive:?false,?autoDelete:?false,?arguments:?queueNormalArgs);
????????????channel.QueueDeclare(queue:?_queueRetry,?durable:?true,?exclusive:?false,?autoDelete:?false,?arguments:?queueRetryArgs);
????????????channel.QueueDeclare(queue:?_queueFail,?durable:?true,?exclusive:?false,?autoDelete:?false,?arguments:?queueFailArgs);

????????????//为队列绑定交换机
????????????channel.QueueBind(queue:?_queueNormal,?exchange:?_exchangeNormal,?routingKey:?"#");
????????????channel.QueueBind(queue:?_queueRetry,?exchange:?_exchangeRetry,?routingKey:?"#");
????????????channel.QueueBind(queue:?_queueFail,?exchange:?_exchangeFail,?routingKey:?"#");

????????????#region?创建一个普通消息消费者
????????????{
????????????????var?consumer?=?new?EventingBasicConsumer(channel);

????????????????consumer.Received?+=?(sender,?e)?=>
????????????????{
????????????????????var?_sender?=?(EventingBasicConsumer)sender;????????????//消息传送者
????????????????????var?_channel?=?_sender.Model;???????????????????????????//消息传送通道
????????????????????var?_message?=?(BasicDeliverEventArgs)e;????????????????//消息传送参数
????????????????????var?_headers?=?_message.BasicProperties.Headers;????????//消息头
????????????????????var?_content?=?Encoding.UTF8.GetString(_message.Body.ToArray());??//消息内容
????????????????????var?_death?=?default(Dictionary<string,?object>);???????//死信参数

????????????????????if?(_headers?!=?null?&&?_headers.ContainsKey("x-death"))
????????????????????????_death?=?(Dictionary<string,?object>)(_headers["x-death"]?as?List<object>)[0];

????????????????????try
????????????????????#region?消息处理
????????????????????{
????????????????????????Console.WriteLine();
????????????????????????Console.WriteLine($"{DateTime.Now.ToString("HH:mm:ss.fff")}\t(1.0)消息接收：\r\n\t[deliveryTag={_message.DeliveryTag}]\r\n\t[consumerID={_message.ConsumerTag}]\r\n\t[exchange={_message.Exchange}]\r\n\t[routingKey={_message.RoutingKey}]\r\n\t[content={_content}]");

????????????????????????throw?new?Exception("模拟消息处理失败效果。");

????????????????????????//处理成功时
????????????????????????Console.WriteLine($"{DateTime.Now.ToString("HH:mm:ss.fff")}\t(1.1)处理成功：\r\n\t[deliveryTag={_message.DeliveryTag}]");

????????????????????????//消息确认?(销毁当前消息)
????????????????????????_channel.BasicAck(deliveryTag:?_message.DeliveryTag,?multiple:?false);
????????????????????}
????????????????????#endregion
????????????????????catch?(Exception?ex)
????????????????????#region?消息处理失败时
????????????????????{
????????????????????????var?retryCount?=?(long)(_death?["count"]????default(long));?//查询当前消息被重新投递的次数?(首次则为0)

????????????????????????Console.WriteLine($"{DateTime.Now.ToString("HH:mm:ss.fff")}\t(1.2)处理失败：\r\n\t[deliveryTag={_message.DeliveryTag}]\r\n\t[retryCount={retryCount}]");

????????????????????????if?(retryCount?>=?2)
????????????????????????#region?投递第3次还没消费成功时，就转发给?exchangeFail?交换机
????????????????????????{
????????????????????????????//消息拒绝（投递给死信交换机，也就是上边定义的?("x-dead-letter-exchange",?_exchangeFail)）
????????????????????????????_channel.BasicNack(deliveryTag:?_message.DeliveryTag,?multiple:?false,?requeue:?false);
????????????????????????}
????????????????????????#endregion
????????????????????????else
????????????????????????#region?否则转发给?exchangeRetry?交换机
????????????????????????{
????????????????????????????var?interval?=?(retryCount?+?1)?*?10;?//定义下一次投递的间隔时间?(单位：秒)

????????????????????????????//定义下一次投递的间隔时间?(单位：毫秒)
????????????????????????????_message.BasicProperties.Expiration?=?(interval?*?1000).ToString();

????????????????????????????//将消息投递给?_exchangeRetry?(会自动增加?death?次数)
????????????????????????????_channel.BasicPublish(exchange:?_exchangeRetry,?routingKey:?_message.RoutingKey,?basicProperties:?_message.BasicProperties,?body:?_message.Body);

????????????????????????????//消息确认?(销毁当前消息)
????????????????????????????_channel.BasicAck(deliveryTag:?_message.DeliveryTag,?multiple:?false);
????????????????????????}
????????????????????????#endregion
????????????????????}
????????????????????#endregion
????????????????};
????????????????channel.BasicConsume(queue:?_queueNormal,?autoAck:?false,?consumer:?consumer);
????????????}
????????????#endregion
????????}

消息延时推送

• 过期队列+死信交换机

• RabbitMQ 3.6.x 开始可以使用延迟插件，交换机类型选择x-delayed-message(延迟将数据放入队列)

???public?static?void?ConsumerMessage()
???{
????????????var?connection?=?RabbitMQHelper.GetConnection();
????????????var?channel?=?connection.CreateModel();

????????????var?exchangeArgumets?=?new?Dictionary<string,?object>
????????????{
????????????????{?"x-delayed-type",?"topic"?}??//延迟交换机的类型
????????????};
????????????channel.ExchangeDeclare("delay_exchange",?"x-delayed-message",?true,?false,?exchangeArgumets);

????????????//?创建队列
????????????string?queueName1?=?"delay_queue1";
????????????channel.QueueDeclare(queueName1,?false,?false,?false,?null);
????????????string?queueName2?=?"delay_queue2";
????????????channel.QueueDeclare(queueName2,?false,?false,?false,?null);
????????????string?queueName3?=?"delay_queue3";
????????????channel.QueueDeclare(queueName3,?false,?false,?false,?null);
????????????//?绑定到交互机
????????????channel.QueueBind(queue:?queueName1,?exchange:?"delay_exchange",?routingKey:?"delayed-direct1");
????????????channel.QueueBind(queue:?queueName2,?exchange:?"delay_exchange",?routingKey:?"delayed-direct2");
????????????channel.QueueBind(queue:?queueName3,?exchange:?"delay_exchange",?routingKey:?"delayed-direct3");

????????????var?properties?=?channel.CreateBasicProperties();
????????????properties.Persistent?=?true;?//?标记消息持久化

????????????Console.WriteLine("?[*]?Waiting?for?messages.");

????????????var?consumer?=?new?EventingBasicConsumer(channel);
????????????consumer.Received?+=?(model,?ea)?=>
????????????{
????????????????var?body?=?ea.Body;
????????????????var?message?=?Encoding.UTF8.GetString(body.ToArray());
????????????????var?routingKey?=?ea.RoutingKey;
????????????????var?_headers?=?ea.BasicProperties.Headers;????????//消息头
????????????????int?delay?=?0;
????????????????if?(_headers?==?null)
????????????????{
????????????????????ea.BasicProperties.Headers?=?new?Dictionary<string,?object>();
????????????????}
????????????????else??if?(?_headers.ContainsKey("x-delay"))
????????????????{
????????????????????delay?=?Convert.ToInt32(ea.BasicProperties.Headers["x-delay"]);
????????????????????delay?=?delay?+?20000;
????????????????}
????????????????ea.BasicProperties.Headers["x-delay"]?=?delay;?//消息头设置消息延迟的时间
????????????????Console.WriteLine($"?{DateTime.Now}=={delay}");
????????????????Console.WriteLine("?[x]?Received?'{0}':'{1}'",?routingKey,?message);
???????????????
????????????????channel.BasicPublish(ea.Exchange,?ea.RoutingKey,?basicProperties:?ea.BasicProperties,?body);
????????????????channel.BasicAck(deliveryTag:?ea.DeliveryTag,?multiple:?false);
????????????};

????????????channel.BasicConsume(queue:?queueName3,
?????????????????????????????????autoAck:?false,
?????????????????????????????????consumer:?consumer);

????????????Console.WriteLine("?Press?[enter]?to?exit.");
????????????Console.ReadLine();
????????}