聊聊RabbitMQ那一些事儿之一基础应用

/// <summary>
 /// 消息生成者
 /// </summary>
public class Program
{
    static void Main(string[] args)
    {
        // rabbitMQ链接对象
        var factory = new ConnectionFactory();
        // RabbitMQ服务在本地运行
        factory.HostName = "192.168.1.1";
        // RabbitMQ服务端口
        factory.Port = 5672;
        // 用户名
        factory.UserName = "guest";
        // 密码
        factory.Password = "guest";
        // 虚拟主机名称
        factory.VirtualHost = "/";

        // 队列名称
        string queueName = "hello";

        // 创建链接
        using (var connection = factory.CreateConnection())
        {
            // 创建通道
            using (var channel = connection.CreateModel())
            {
                // 创建一个名称为hello的消息队列--当然一步也可以通过RabbitMQ管理后台添加
                // 当已经存在该队列时，不会重复添加，但是如果已存在的队列和新建的队列存在属性差异时，会创建失败，会抛异常，所以在实际使用时，如果要通过程序创建队列，最好要捕捉异常，避免因为这样的问题而导致程序崩溃。
                channel.QueueDeclare(queueName, false, false, false, null);
                Console.WriteLine("我是生成者");

                while (true)
                {
                    Console.WriteLine("请输入你要发送的消息,并按Enter键结束");

                    // 接收用户输入的消息
                    string message = Console.ReadLine();
                    // 消息编码
                    var body = Encoding.UTF8.GetBytes(message);
                    // 向消息服务器推送消息
                    channel.BasicPublish("", queueName, null, body);

                    Console.WriteLine($"已发送 {System.DateTime.Now.ToString("HH:mm:ss")}： {message}");
                }
            }
        }
    }
}

 /// <summary>
 /// 消息消费者
 /// </summary>
 public class Program
 {
     static void Main(string[] args)
     {
         // rabbitMQ链接对象
         var factory = new ConnectionFactory();
         // RabbitMQ服务在本地运行
         factory.HostName = "192.168.1.1";
         // RabbitMQ服务端口
         factory.Port = 5672;
         // 用户名
         factory.UserName = "guest";
         // 密码
         factory.Password = "guest";
         // 虚拟主机名称
         factory.VirtualHost = "/";

         // 队列名称
         string queueName = "hello";

         // 创建链接
         using (var connection = factory.CreateConnection())
         {
             // 创建通道
             using (var channel = connection.CreateModel())
             {

                 // 创建一个名称为hello的消息队列--当然一步也可以通过RabbitMQ管理后台添加
                 // 当已经存在该队列时，不会重复添加，但是如果已存在的队列和新建的队列存在属性差异时，会创建失败，会抛异常，所以在实际使用时，如果要通过程序创建队列，最好要捕捉异常，避免因为这样的问题而导致程序崩溃。
                 channel.QueueDeclare(queueName, false, false, false, null);
                 Console.WriteLine("我是消费者");

                 // 创建一个消费者
                 var consumer = new EventingBasicConsumer(channel);
                 // 订阅对应的消息 autoAck:是否自动确认
                 channel.BasicConsume(queueName, autoAck:false, consumer);

                 consumer.Received += (model, ea) =>
                 {
                     var body = ea.Body;
                     var message = Encoding.UTF8.GetString(body);
                     Console.WriteLine($"已接收 {System.DateTime.Now.ToString("HH:mm:ss")}： {message}");

                     // 为了模拟推送过程，在此程序休息1分钟
                     Thread.Sleep(6000);
                     // 确认消费
                     channel.BasicAck(ea.DeliveryTag, false);
                 };
                 Console.ReadLine();
             }
         }
     }
 }

 // 通过Qos设置每次接收消息的条数
 // 三个参数说明
 // prefetchSize：为预取的长度，一般设置为0即可，表示长度不限
 // prefetchCount：表示预取的条数，即发送的最大消息条数
 // global表示是否在Connection中全局设置，true表示Connetion下的所有channel都设置为这个配置。
 channel.BasicQos(prefetchSize: 0,
                  prefetchCount: 1,
                  global: false);