RabbitMQ

 介绍

• Producer：消息的生产者（发送消息的程序）。
• Queue：消息队列，理解为一个容器，生产者向它发送消息，它把消息存储，等待消费者消费。
• Consumer：消息的消费者（接收消息的程序）。
• 此处我们假设 Consumer1、Consumer2、Consumer3 分别为完成任务速度不一样快的消费者，这会引出此模式的一个重点问题。

　　如何理解：

　　　　工作模式由图可以看出，就是在简单队列模式的基础上，增加了多个消费者，也就是让多个消费者绑定同一个队列，共同去消费，这样能解决简单队列模式中，如果生产速速远大于消费速度，而导致的消息堆积现象。

• 因为消息被消费后就会消失，所以不必担心任务会重复执行。

 　　代码实现：

　　　　注：工作队列模式有两种

1. 轮询模式：每个消费者均分消息
2. 公平分发模式（能者多劳）：按能力分发，处理速度快的分发的多，处理速度慢的分发的少

 Worker模式

 　　Worker模式其实就是一对多模式，我们定义两个消费者来看看效果：

 　　默认情况下，RabbitMQ会顺序的将message发给下一个消费者。每个消费者会得到平均数量的message。这种方式称之为round-robin（轮询)。

　　但是很多情况下并不希望消息平均分配,而是要消费快的多消费,消费少的少消费.还有很多情况下一旦其中一个宕机,那么另外接收者的无法接收原本这个接收者所要接收的数据。

　　我们修改其中一个消费者代码,让其等待5秒。在等待中停止运行 看看效果。

consumer.Received += (ch, ea) =>
{
    string message = Encoding.Default.GetString(ea.Body.ToArray());
    Console.WriteLine($"接受到消息：{message}");
    Thread.Sleep(5000);
};

 

 　　当消费者宕机后消费者1并没有接受宕机后的数据。所以我们需要消息确认来解决这个问题。

 RabbitMQ消息确认

 Rabbit中存在两种消息确认模式

•  自动模式 - 只要消息从队列获取,无论消费者获取到消息后是否成功消费,都认为是消息成功消费。
•  手动模式 - 消费从队列中获取消息后,服务器会将该消息处于不可用状态,等待消费者反馈。如果消费者在消费过程中出现异常，断开连接且没有发送应答，那么RabbitMQ会将这个消息重新投递。

　　也就是说我们只要将消息确认模式改为手动即可,改为手动确认方式只需改两处,1.开启监听时将autoAck参数改为false,2.消息消费成功后返回确认

 修改两个消费者代码,并在其中一个中延迟确认。

consumer.Received += (model, ea) =>
{
    Thread.Sleep(1000);//等待1秒,
    byte[] message = ea.Body;//接收到的消息
    Console.WriteLine("接收到信息为:" + Encoding.UTF8.GetString(message));
    //返回消息确认
    channel.BasicAck(ea.DeliveryTag, true);
};
//消费者开启监听
//将autoAck设置false 关闭自动确认
channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);

 

 

　　如果在延迟中消费者断开连接,那么RabbitMQ会重新投递未确认的消息

 能者多劳模式

　　能者多劳是给消费速度快的消费更多的消息.少的责消费少的消息.能者多劳是建立在手动确认基础上实现。
　　在延迟确认的消费中【消费者】添加BasicQos

// 3、创建通道
IModel channel = connection.CreateModel(); 
// 3.1、每次只能向消费者发送一条信息,再消费者未确认之前,不再向他发送信息
channel.BasicQos(0, 1, false);

 

 　　可以看出消费快的消费者接受了更多的消息,这就是能者多劳模式的体现

 完整测试代码：

•  生产者实现

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();
        }
    }
}

•  消费者实现1

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(); 
            // 3.1、每次只能向消费者发送一条信息,再消费者未确认之前,不再向他发送信息
            channel.BasicQos(0, 1, false);
            // 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}");
                  Thread.Sleep(5000);
                  // 6、确认该消费已被消费
                  channel.BasicAck(ea.DeliveryTag, true);
              };
            // 7、启动消费者 设置为自动应答消息
            channel.BasicConsume(
                queue: name, //消息队列名称
                autoAck: false, //两种消息确认模式false 手动模式 true自动模式
                consumer: consumer);

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

•  消费者实现2

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

namespace RabbitMQTest.Consumer2
{
    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}");
                // 6、确认该消费已被消费
                channel.BasicAck(ea.DeliveryTag, true);
            };
            // 7、启动消费者 设置为自动应答消息
            channel.BasicConsume(name, false, consumer);

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