RabbitMQ

 介绍

 

• Producer：消息的生产者（发送消息的程序）。
• Queue：消息队列，理解为一个容器，生产者向它发送消息，它把消息存储，等待消费者消费。
• Consumer：消息的消费者（接收消息的程序）。

　　由图所示，简单队列模式，一个生产者，经过一个队列，对应一个消费者。可以看做是点对点的一种传输方式，相较架构模型图，最主要的特点就是看不到 Exchange（交换机） 和 routekey（路由键） ，正是因为这种模式简单，所以并不会涉及到复杂的条件分发等等，因此也不需要用户去显式的考虑交换机和路由键的问题。

• 但是要注意，这种模式并不是生产者直接对接队列，而是用了默认的交换机，默认的交换机会把消息发送到和 routekey 名称相同的队列中去，这也是我们在后面代码中在 routekey 位置填写了队列名称的原因

 .net 5.0 NuGet引用包

 RabbitMQ.Client

 简单测试代码：

•  生产者实现

using RabbitMQ.Client;
using System;
using System.Text;

namespace RabbitMQTest.Producer
{
    /// <summary>
    /// RabbitMQ测试_Producer生产者
    /// </summary>
    class Program
    {
        static void Main(string[] args)
        {
            // 1、创建连接工厂
            IConnectionFactory factory = new ConnectionFactory()
            {
                UserName = "admin",
                Password = "admin",
                HostName = "192.168.1.101", //rabbitmq server ip
                Port = 5672,
                VirtualHost = "testhost" //在此连接期间要访问的虚拟主机。
            };
            // 2、创建连接
            IConnection connection = factory.CreateConnection();
            // 3、创建通道
            IModel channel = connection.CreateModel();
            string name = "testQueue";
            // 4、声明一个队列
            channel.QueueDeclare(
                queue: name, //消息队列名称
                durable: false, //是否持久化，true持久化，队列会保存磁盘,服务器重启时可以保证不丢失相关信息。
                exclusive: false, //是否排他，true排他的，如果一个队列声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除。
                autoDelete: false, //是否自动删除，true是自动删除，自动删除的前提是：致少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。
                arguments: null); //设置队列的一些其它参数。

            Console.WriteLine("
 RabbitMQ连接成功，请输入消息，输入exit退出！");
            string input;
            do
            {
                input = Console.ReadLine();
                byte[] sendBytes = Encoding.UTF8.GetBytes(input);
                //发布消息
                channel.BasicPublish("", name, null, sendBytes);
                Console.WriteLine("消息发布完毕");
            } while (input.Trim().ToLower() != "exit");
            Console.WriteLine("
 RabbitMQ测试完毕！");
            // 6、关闭通道
            channel.Close();
            // 7、关闭连接
            connection.Close();
        }
    }
}

• queueDeclare 方法解释
  •  参数1：queue（队列名称），如果队列不存在，则自动创建。
  •  参数2：durable（队列是否持久化），持久化可以保证服务器重启后此队列仍然存在。
  •  参数3：exclusive（排他队列）即是否独占队列，如果此项为 true，该队列仅对首次申明它的连接可见，并在连接断开时自动删除。
  •  参数4：autoDelete（自动删除），最后一个消费者将消息消费完毕后，自动删除队列。
  •  参数5：arguments（携带附加属性）。
• basicPublish 方法解释
  •  参数1：exchange（交换机名称）。
  •  参数2：routingKey（路由key），此处填写队列名，可理解为把消息发送到和 routekey 名称相同的队列中去。
  •  参数3：props（消息的控制状态），可以在此处控制消息的持久化。
    • 参数为：MessageProperties.PERSISTENT_TEXT_PLAIN
  • 参数4：body（消息主体），类型是一个字节数组，要转一下类型。

通过工具关闭channel和释放连接：先关闭通道，再释放连接。

• 消费者实现

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
using System.Threading;

namespace RabbitMQTest.Consumer
{
    /// <summary>
    /// RabbitMQ测试_Consumer消费者
    /// </summary>
    class Program
    {
        static void Main(string[] args)
        {
            // 1、创建连接工厂
            ConnectionFactory factory = new ConnectionFactory()
            { 
                UserName = "admin", //用户名
                Password = "admin", //密码
                HostName = "192.168.1.101", //rabbitmq server ip
                Port = 5672, //端口号
                VirtualHost = "testhost" //在此连接期间要访问的虚拟主机。

            };
            // 2、创建连接
            IConnection connection = factory.CreateConnection();
            // 3、创建通道
            IModel channel = connection.CreateModel();
            // 4、事件基本消费者
            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
            string name = "testQueue";
            // 5、接收到消息事件
            consumer.Received += (ch, ea) =>
              {
                  string message = Encoding.Default.GetString(ea.Body.ToArray());
                  Console.WriteLine($"接受到消息：{message}");
                  Console.WriteLine($"收到该消息[{ea.DeliveryTag}] 延迟1s发送回执！");
                  Thread.Sleep(1000);
                  // 6、确认该消费已被消费
                  channel.BasicAck(ea.DeliveryTag, false);
                  Console.WriteLine($"已发送回执{ea.DeliveryTag}");
              };
            // 7、启动消费者 设置为手动应答消息
            channel.BasicConsume(name, false, consumer);

            Console.WriteLine("消费者已启动");
            Console.ReadKey();
            channel.Close();
            connection.Close();
        }
    }
}

• basicConsume 方法解释
  • 参数1：queue（队列名称），即消费哪个队列的消息 。
  • 参数2：autoAck（自动应答）开始消息的自动确认机制，只要消费了就从队列删除消息。
  • 参数3：callback（消费时的回调接口），callback 的类型是 Consumer 这里使用了 DefaultConsumer 就是 Consumer 的一个实现类。其中重写 handleDelivery 方法，就可以获取到消费的数据内容了，这里主要使用了其中的 body，即查看消息主体，其他三个参数暂时还没用到，有兴趣可以先打印输出一下，能先有个大概的了解。