在C#中使用RabbitMQ

官方提供了 6 中应用场景使用教程： https://www.rabbitmq.com/getstarted.html

使用消息队列的好处：

业务系统往往要求相应能力特别强，起到削峰填谷的作用。
解耦和高可用。如果一个系统挂了，不会影响到其他系统的运行。
业务系统往往有对消息的高可靠要求，以及有对复杂功能（如ACK）的要求。
增强业务系统的异步处理能力，减少甚至几乎不可能出现并发现象。

该案例使用 RabbitMQ.Client 包

基本用法

  <appSettings>
    <add key="AppID" value="150107"/>
    <add key="RabbitMQUri" value="amqp://test:123456@localhost:5672" />
  </appSettings>

/// <summary>
/// 发送消息
/// </summary>
public class Send
{
    private static readonly string appID = ConfigurationManager.AppSettings["AppID"];

    static void Main(string[] args)
    {
        var factory = new ConnectionFactory { Uri = ConfigurationManager.AppSettings["RabbitMQUri"] };
        using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                string queue = string.Format("MQ{0}.BaseStudy", appID);

                channel.QueueDeclare(queue, false, false, false, null);   //定义一个队列

                while (true)
                {
                    Console.Write("请输入要发送的消息：");
                    var message = Console.ReadLine();
                    var body = Encoding.UTF8.GetBytes(message);

                    channel.BasicPublish("", queue, null, body); //发送消息

                    Console.WriteLine("已发送的消息： {0}", message);
                }
            }
        }            
    }
}

/// <summary>
/// 接收消息
/// </summary>
public class Receive
{
    private static readonly string appID = ConfigurationManager.AppSettings["AppID"];

    static void Main(string[] args)
    {
        var factory = new ConnectionFactory { Uri = ConfigurationManager.AppSettings["RabbitMQUri"] };
        using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                string queue = string.Format("MQ{0}.BaseStudy", appID);

                channel.QueueDeclare(queue, false, false, false, null);   //定义一个队列

                Console.WriteLine("准备接收消息：");                    

                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (s, e) =>
                {
                    var body = e.Body;
                    var message = Encoding.UTF8.GetString(body);
                    Console.WriteLine("接收到的消息： {0}", message);
                };
                channel.BasicConsume(queue, true, consumer);  //开启消费者与通道、队列关联

                Console.ReadLine();
            }
        }            
    }
}

主题发布订阅

/// <summary>
/// 发送消息，采用主题模式
/// </summary>
public class Send
{
    private static readonly string appID = ConfigurationManager.AppSettings["AppID"];

    static void Main(string[] args)
    {
        var factory = new ConnectionFactory { Uri = ConfigurationManager.AppSettings["RabbitMQUri"] };
        using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                string exchange = string.Format("Ex{0}.Logs", appID);

                channel.ExchangeDeclare(exchange, "topic");    //声明创建一个交换机，交换机类型设定为‘topic’

                while (true)
                {
                    Console.Write("请输入要发送的消息，输入格式如'RoutingKey_Message'：");
                    var keyWithMsg = Console.ReadLine();

                    args = keyWithMsg.Split('_');
                    var routingKey = args.Length > 1 ? args[0] : "*.rabbit";
                    var message = args.Length > 1 ? args[1] : "Hello World";
                    var body = Encoding.UTF8.GetBytes(message);

                    channel.BasicPublish(exchange, routingKey, null, body);    //发布消息

                    Console.WriteLine("已发送的消息： '{0}':'{1}'", routingKey, message);
                }
            }
        }            
    }
}

/// <summary>
/// 接收消息，采用主题模式
/// </summary>
public class Receive
{
    private static readonly string appID = ConfigurationManager.AppSettings["AppID"];

    static void Main(string[] args)
    {
        var factory = new ConnectionFactory { Uri = ConfigurationManager.AppSettings["RabbitMQUri"] };
        using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                string exchange = string.Format("Ex{0}.Logs", appID);                    

                channel.ExchangeDeclare(exchange, "topic");    //声明创建一个交换机，交换机类型设定为‘topic’
                var queueName = channel.QueueDeclare().QueueName;   //获取连接通道所使用的队列

                Console.Write("请输入准备监听的消息主题格式，格式如'*.rabbit'或者'info.*'或者'info.warning.error'等：");

                while (true)
                {
                    var bindingKey = Console.ReadLine();
                    channel.QueueBind(queueName, exchange, bindingKey);   //队列绑定到交换机

                    Console.WriteLine("准备接收消息");

                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (s, e) =>
                    {
                        var routingKey = e.RoutingKey;
                        var body = e.Body;
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine("接收到的消息： '{0}':'{1}'", routingKey, message);
                    };
                    channel.BasicConsume(queueName, true, consumer);    //开启消费者与通道、队列关联
                }
            }
        }
    }
}

高级特性

/// <summary>
/// 发送消息
/// </summary>
public class Send
{
 private static readonly string appID = ConfigurationManager.AppSettings["AppID"];

 static void Main(string[] args)
 {
     var factory = new ConnectionFactory { Uri = ConfigurationManager.AppSettings["RabbitMQUri"] };
     using (var connection = factory.CreateConnection())
     {
         using (var channel = connection.CreateModel())
         {
             string queue = string.Format("MQ{0}.TaskQueue", appID);

             channel.QueueDeclare(queue, true, false, false, null);   //定义一个支持持久化的消息队列
             var properties = channel.CreateBasicProperties();
             properties.DeliveryMode = 2;    //1表示不持久，2表示持久化

             Console.WriteLine("请注意：演示耗时较长的消息时，可通过发送带有‘.’的内容去模拟，每个‘.’加1秒！");

             while (true)
             {
                 Console.Write("请输入要发送的消息：");
                 var message = Console.ReadLine();
                 var body = Encoding.UTF8.GetBytes(message);

                 channel.BasicPublish("", queue, properties, body);   //发送消息

                 Console.WriteLine("已发送的消息： {0}", message);
             }
         }
     }
 }
}

/// <summary>
/// 接收消息
/// </summary>
public class Receive
{
 private static readonly string appID = ConfigurationManager.AppSettings["AppID"];

 static void Main(string[] args)
 {
     var factory = new ConnectionFactory { Uri = ConfigurationManager.AppSettings["RabbitMQUri"] };
     using (var connection = factory.CreateConnection())
     {
         using (var channel = connection.CreateModel())
         {
             string queue = string.Format("MQ{0}.TaskQueue", appID);

             channel.QueueDeclare(queue, true, false, false, null);   //定义一个支持持久化的消息队列

             channel.BasicQos(0, 1, false);  //在一个消费者还在处理消息且没响应消息之前，不要给他分发新的消息，而是将这条新的消息发送给下一个不那么忙碌的消费者

             Console.WriteLine("准备接收消息：");
             var consumer = new EventingBasicConsumer(channel);
             consumer.Received += (s, e) =>
             {
                 var message = Encoding.UTF8.GetString(e.Body);
                 SimulationTask(message);

                 channel.BasicAck(e.DeliveryTag, false); //手动Ack：用来确认消息已经被消费完成了
             };
             channel.BasicConsume(queue, false, consumer);    //开启消费者与通道、队列关联
             //channel.BasicConsume(queue, true, consumer);    //开启消费者与通道、队列关联；自动Ack

             Console.ReadLine();
         }
     }            
 }

 /// <summary>
 /// 模拟消息任务的处理过程
 /// </summary>
 /// <param name="message">消息</param>
 private static void SimulationTask(string message)
 {
     Console.WriteLine("接收的消息： {0}", message);
     int dots = message.Split('.').Length - 1;
     Thread.Sleep(dots * 1000);  
     Console.WriteLine("接收的消息处理完成，现在时间为{0}！", DateTime.Now);            
 }
}