RabbitMQ任务分发机制

1. Message acknowledgment 消息确认

每个Consumer可能需要一段时间才能处理完收到的数据。如果在这个过程中,Consumer出错了,异常退出了,而数据还没有处理完成,那么
非常不幸,这段数据就丢失了。因为我们采用no-ack的方式进行确认,也就是说,每次Consumer接到数据后,而不管是否处理完
成,RabbitMQ Server会立即把这个Message标记为完成,然后从queue中删除了。

    
如果一个Consumer异常退出了,它处理的数据能够被另外的Consumer处理,这样数据在这种情况下就不会丢失了(注意是这种情况下)。

     
为了保证数据不被丢失,RabbitMQ支持消息确认机制,即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到,那么我们不能采用no-ack。而应该是在处理完数据后发送ack。

   
在处理数据后发送的ack,就是告诉RabbitMQ数据已经被接收,处理完成,RabbitMQ可以去安全的删除它了。

   
如果Consumer退出了但是没有发送ack,那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。

   
这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说,RabbitMQ给了Consumer足够长的时间来做数据处理。

这样即使你通过Ctr-C中断了Recieve.cs,那么Message也不会丢失了,它会被分发到下一个Consumer。

     
如果忘记了ack,那么后果很严重。当Consumer退出时,Message会重新分发。然后RabbitMQ会占用越来越多的内存,由于
RabbitMQ会长时间运行,因此这个“内存泄漏”是致命的。去调试这种错误,可以通过一下命令打印un-acked
Messages.

当有Consumer需要大量的运算时,RabbitMQ
Server需要一定的分发机制来balance每个Consumer的load。接下来我们分布讲解。 

2. Round-robin dispatching 循环分发

       
RabbitMQ的分发机制非常适合扩展,而且它是专门为并发程序设计的。如果现在load加重,那么只需要创建更多的Consumer来进行任务处理即
可。当然了,对于负载还要加大怎么办?我没有遇到过这种情况,那就可以创建多个virtual
Host,细化不同的通信类别了。

     1、首先开启两个Consumer,即运行两个Recieve.cs。

     2、在开启两个Producer,即运行两个Producer.cs。

默认情况下,RabbitMQ
会顺序的分发每个Message。当每个收到ack后,会将该Message删除,然后将下一个Message分发到下一个Consumer。这种分发方式叫做round-robin(优雅分发)。

Producer.cs

class Program
    {
        static void Main(string[] args)
        {
            ConnectionFactory factory = new ConnectionFactory() { HostName = "localhost" };
            using (IConnection connection = factory.CreateConnection())
            {
                using (IModel channel = connection.CreateModel())
                {
                    channel.QueueDeclare("hello", false, false, false, null);
                    var message = GetMessage(args);
                    var body = Encoding.UTF8.GetBytes(message);

                    var properties = channel.CreateBasicProperties();
                    properties.DeliveryMode = 2;//non-persistent (1) or persistent (2) 
                    //channel.TxSelect();
                    channel.BasicPublish("", "hello", properties, body);
                    //channel.TxCommit();
                }
            }
        }

        private static string GetMessage(string[] args)
        {
            return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!");
        }
    }

Consumer.cs

//#define demo1
#define demo2
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ReceiveDemo2
{
    /// <summary>
    /// 一个Send和多个Receive的例子,
    /// 还加上了ack的例子.
    /// 优雅分发
    /// </summary>
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("hello", false, false, false, null);
                    var consumer = new QueueingBasicConsumer(channel);
#if demo1
                    channel.BasicConsume("hello", true, consumer);//自动删除消息
#else
                    channel.BasicConsume("hello", false, consumer);//需要接受方发送ack回执,删除消息
#endif
                    Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C");
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//挂起的操作
#if demo2
                        channel.BasicAck(ea.DeliveryTag, false);//与channel.BasicConsume("hello", false, null, consumer);对应
#endif
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine(" [x] Received {0}", message);
                        int dots = message.Split('.').Length - 1;
                        Thread.Sleep(dots * 1000);
                        Console.WriteLine(" [x] Done");
#if demo2
                        //channel.BasicAck(ea.DeliveryTag, false);//与channel.BasicConsume("hello", false, null, consumer);对应,这句话写道40行和49行运行结果就会不一样.写到这里会发生如果输出[x] Received {0}之后,没有输出 [x] Done之前,CTRL+C结束程序,那么message会自动发给另外一个客户端,当另外一个客户端收到message后,执行完49行的命令之后,服务器会删掉这个message
#endif
                    }
                }
            }
        }
    }
}

2. Round-robin dispatching 循环分发

       
RabbitMQ的分发机制非常适合扩展,而且它是专门为并发程序设计的。如果现在load加重,那么只需要创建更多的Consumer来进行任务处理即
可。当然了,对于负载还要加大怎么办?我没有遇到过这种情况,那就可以创建多个virtual
Host,细化不同的通信类别了。

     1、首先开启两个Consumer,即运行两个Recieve.cs。

     2、在开启两个Producer,即运行两个Producer.cs。

默认情况下,RabbitMQ
会顺序的分发每个Message。当每个收到ack后,会将该Message删除,然后将下一个Message分发到下一个Consumer。这种分发方式叫做round-robin(优雅分发)。

Producer.cs

class Program
    {
        static void Main(string[] args)
        {
            ConnectionFactory factory = new ConnectionFactory() { HostName = "localhost" };
            using (IConnection connection = factory.CreateConnection())
            {
                using (IModel channel = connection.CreateModel())
                {
                    channel.QueueDeclare("hello", false, false, false, null);
                    var message = GetMessage(args);
                    var body = Encoding.UTF8.GetBytes(message);

                    var properties = channel.CreateBasicProperties();
                    properties.DeliveryMode = 2;//non-persistent (1) or persistent (2) 
                    //channel.TxSelect();
                    channel.BasicPublish("", "hello", properties, body);
                    //channel.TxCommit();
                }
            }
        }

        private static string GetMessage(string[] args)
        {
            return ((args.Length > 0) ? string.Join(" ", args) : "Hello World!");
        }
    }

Consumer.cs

//#define demo1
#define demo2
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ReceiveDemo2
{
    /// <summary>
    /// 一个Send和多个Receive的例子,
    /// 还加上了ack的例子.
    /// 优雅分发
    /// </summary>
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("hello", false, false, false, null);
                    var consumer = new QueueingBasicConsumer(channel);
#if demo1
                    channel.BasicConsume("hello", true, consumer);//自动删除消息
#else
                    channel.BasicConsume("hello", false, consumer);//需要接受方发送ack回执,删除消息
#endif
                    Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C");
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//挂起的操作
#if demo2
                        channel.BasicAck(ea.DeliveryTag, false);//与channel.BasicConsume("hello", false, null, consumer);对应
#endif
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine(" [x] Received {0}", message);
                        int dots = message.Split('.').Length - 1;
                        Thread.Sleep(dots * 1000);
                        Console.WriteLine(" [x] Done");
#if demo2
                        //channel.BasicAck(ea.DeliveryTag, false);//与channel.BasicConsume("hello", false, null, consumer);对应,这句话写道40行和49行运行结果就会不一样.写到这里会发生如果输出[x] Received {0}之后,没有输出 [x] Done之前,CTRL+C结束程序,那么message会自动发给另外一个客户端,当另外一个客户端收到message后,执行完49行的命令之后,服务器会删掉这个message
#endif
                    }
                }
            }
        }
    }
}

3. Message durability消息持久化

    
在上一节中我们知道了即使Consumer异常退出,Message也不会丢失。但是如果RabbitMQ
Server退出呢?软件都有bug,即使RabbitMQ
Server是完美毫无bug的(当然这是不可能的,是软件就有bug,没有bug的那不叫软件),它还是有可能退出的:被其它软件影响,或者系统重启
了,系统panic了。。。

   
为了保证在RabbitMQ退出或者crash了数据仍没有丢失,需要将queue和Message都要持久化。queue的持久化需要在声明时指定durable=True,修改Producer和Consumer的channel.QueueDeclare代码,再次强调,Producer和Consumer都应该去创建这个queue,尽管只有一个地方的创建是真正起作用的:

bool durable = true;
channel.QueueDeclare("hello", durable, false, false, null);

上述语句执行不会有什么错误,但是确得不到我们想要的结果,原因就是RabbitMQ
Server已经维护了一个叫hello的queue,那么上述执行不会有任何的作用,也就是hello的任何属性都不会被影响。这一点在上篇文章也讨论过。

那么workaround也很简单,声明一个另外的名字的queue,比如名字定位task_hello,或者通过监控,删除名为“hello”的Queue。

接下来,还需要持久化Message,即在Producer.cs里面Publish的时候指定一个properties,方式如下:

static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    bool durable = true;
                    channel.QueueDeclare("task_queue", durable, false, false, null);//queue的持久化需要在声明时指定durable=True
                    var message = GetMessage(args);
                    var body = Encoding.UTF8.GetBytes(message);
                    var properties = channel.CreateBasicProperties();
                    properties.SetPersistent(true);//需要持久化Message,即在Publish的时候指定一个properties,
                    channel.BasicPublish("", "task_hello", properties, body);
                }
            }
        }

关于持久化的进一步讨论:

    为了数据不丢失,我们采用了:

    但是这样能保证数据100%不丢失吗?

   
答案是否定的。问题就在与RabbitMQ需要时间去把这些信息存到磁盘上,这个time
window虽然短,但是它的确还是有。在这个时间窗口内如果数据没有保存,数据还会丢失。还有另一个原因就是RabbitMQ并不是为每个Message都做fsync:它可能仅仅是把它保存到Cache里,还没来得及保存到物理磁盘上。

   
因此这个持久化还是有问题。但是对于大多数应用来说,这已经足够了。当然为了保持一致性,你可以把每次的publish放到一个transaction中。这个transaction的实现需要user
defined codes。

   
那么商业系统会做什么呢?一种可能的方案是在系统panic时或者异常重启时或者断电时,应该给各个应用留出时间去flash
cache,保证每个应用都能exit gracefully。

1. Message acknowledgment 消息确认

每个Consumer可能需要一段时间才能处理完收到的数据。如果在这个过程中,Consumer出错了,异常退出了,而数据还没有处理完成,那么
非常不幸,这段数据就丢失了。因为我们采用no-ack的方式进行确认,也就是说,每次Consumer接到数据后,而不管是否处理完
成,RabbitMQ Server会立即把这个Message标记为完成,然后从queue中删除了。

    
如果一个Consumer异常退出了,它处理的数据能够被另外的Consumer处理,这样数据在这种情况下就不会丢失了(注意是这种情况下)。

     
为了保证数据不被丢失,RabbitMQ支持消息确认机制,即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到,那么我们不能采用no-ack。而应该是在处理完数据后发送ack。

   
在处理数据后发送的ack,就是告诉RabbitMQ数据已经被接收,处理完成,RabbitMQ可以去安全的删除它了。

   
如果Consumer退出了但是没有发送ack,那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。

永利集团304com,   
这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说,RabbitMQ给了Consumer足够长的时间来做数据处理。

这样即使你通过Ctr-C中断了Recieve.cs,那么Message也不会丢失了,它会被分发到下一个Consumer。

     
如果忘记了ack,那么后果很严重。当Consumer退出时,Message会重新分发。然后RabbitMQ会占用越来越多的内存,由于
RabbitMQ会长时间运行,因此这个“内存泄漏”是致命的。去调试这种错误,可以通过一下命令打印un-acked
Messages.

发表评论

电子邮件地址不会被公开。 必填项已用*标注