WindySky / 应用场景 / RabbitMq应用二

0 0

   

RabbitMq应用二

2017-07-28  WindySky

在应用一中,基本的消息队列使用已经完成了,在实际项目中,一定会出现各种各样的需求和问题,rabbitmq内置的很多强大机制和功能会帮助我们解决很多的问题,下面就一个一个的一起学习一下。

消息响应机制

应用一的列子,在消费者从指定队列获取消息的时候,把通知参数no_ack给设成true了,这样就不需要给rabbitMq服务发送已经处理完毕的通知,rabbitmq把消息发出去后,就会直接删除掉,不去管消费者是否处理成功,这样在实际项目中存在很大的风险,出现代码的健壮性很差的错误。所以一定要把no_ack参数设成false:

1
channel.BasicConsume("newQueue", false, customer);

在接受逻辑全部处理成功后加上一句代码,通知rabbitmq,接到通知后才会删除

复制代码
 var ea = (BasicDeliverEventArgs)customer.Queue.Dequeue();
                        //将消息二进制转回字符串
                        var msg = Encoding.UTF8.GetString(ea.Body);
                        //通知队列,已经处理完成
                        channel.BasicAck(ea.DeliveryTag, false);
                        Console.WriteLine(msg);
复制代码

 

消息持久化

响应保证了消息不会被错误删除,假如rabbitmq挂了,所有消息全部会丢掉,rabbitmq一个广泛使用的机制就是可以持久化,做持久化要两步

1.队列持久化

 //队列是否持久化
                    bool durable = true;
channel.QueueDeclare("firstQueue",durable,false,false,null);

2.消息持久化,通过设置IBasicProperties.SetPersistent来做

 //消息持久化
                    var properties = channel.CreateBasicProperties();
                    properties.SetPersistent(true);
                    properties.DeliveryMode = 2; //消息是持久的,存在并不会受服务器重启影响  

上面的持久化,大部分时候不会出现问题,但是假如在写入队列的时候rabbitmq挂了,还是不会持久上,这种情况,我们就要用到我们代码的逻辑来强制进行持久化了。。。。

负载均衡分发消息

如果有两个接收端消费者同时订阅一个队列,会出现不固定的分发流程,某个消费者可能会出现过多的消息流入造成压力,而另一个空闲的蛋疼。所以,如果能公平的接受消息,处理完一个,接受另一个,同时保证压力的均衡。代码在消费者端设置:

channel.BasicQos(0, 1, false);

上面是几个rabbitmq比较重要的机制,下面开始是rabbitmq的核心牛逼的东西路由

这里涉及2个概念:

1.exchange,这是交换机,也叫路由器,在消息生产者发送消息的时候,实际上不是直接发送到queue队列中,因为他不知道发送到哪个队列,他会先发送到路由器中exchange里,exchange再通过路由匹配把消息发送到匹配的队列当中。

 

2.routingKey这个是路由的匹配规则,当消息发送到exchange里后,会根据routingkey来匹配到底发送到哪个队列,如果没匹配到,则消息丢失

 

exchange的四种类型:

 

1.direct:按routingkey的名称匹配

 

2.fanout:广播,无需匹配routingkey消息会发送到所有队列

 

3.topic:这个是贪婪匹配,也是最灵活的匹配方式,有两种符号#,*.,......*号的意思是

 

 #符号的意思是比如a_#,可以匹配的队列可以是a_a,a_aa,a_aaaaaa,a_a_b.......多词

 

 *符号的意识是比如a_*,可以匹配的队列可以是a_a,a_b,a_c.......单词

 

这个是应用一中发送消息给队列的代码,

 

 

channel.BasicPublish("", "firstQueue", null, body);

 

 

通过查看这个方法的参数中可看到第一个参数是exchange路由,第二个是routingkey匹配规则,而发送的代码第一个参数是"",第二个参数是firstQueue,开始以为是队列实际并不是,原因是如果用空字符串去申明一个exchange,那么系统就会使用"amq.direct"这个exchange。我们在创建一个queue的时候,默认的都会有一个和新建queue同名的routingKey绑定到这个默认的exchange上去,因为在第一个参数选择了默认的exchange,而我们申明的队列叫firstQueue,所以默认的,它在新建一个也叫firstQueue的routingKey,并绑定在默认的exchange上,导致了我们可以在第二个参数routingKey中写firstQueue,这样它就会找到定义的同名的queue,并把消息放进去。

------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

基本概念已经差不多了,我是很擅长排版解释,往下是各种匹配规则的代码和运行情况,直接上代码:

1.路由类型direct,匹配规则rroutingKey相同,一个生产者,两个消费者,采用负载均衡方式分发:

生产者

 

复制代码
 //创建链接工厂,设置目标,用户,密码
            var factory = new ConnectionFactory() { 
                HostName = "127.0.0.1",
                UserName = "feiyang",
                Password = "123456",
                AutomaticRecoveryEnabled = true, //自动重连 
                RequestedHeartbeat = UInt16.MaxValue//心跳超时时间
            };
            
            //开启当前服务设置的用户的链接
            using (var connection =  factory.CreateConnection())
            {
                //开启一个频道
                using (var channel = connection.CreateModel())
                {
                    //创建一个队列
                    //队列是否持久化
                    bool durable = true;
                    //已经存在的队列,不能再定义持久化
                   // channel.QueueDeclare("firstQueue",false,false,false,null);
                    //创建一个新的,持久的交换区  
                    channel.ExchangeDeclare("NewExchange", ExchangeType.Direct, true, false, null);  
                    //持久的队列, 没有排他性,与不自动删除 
                    channel.QueueDeclare("newQueue", durable, false, false, null);
                    // 绑定队列到交换区  
                    channel.QueueBind("newQueue", "NewExchange", "newRoutingKey"); 
                    //消息持久化
                    var properties = channel.CreateBasicProperties();
                    properties.SetPersistent(true);
                    properties.DeliveryMode = 2; //消息是持久的,存在并不会受服务器重启影响  
                    byte[] body = null;
                    //消息是以二进制数组的形式传输的,所以如果消息是实体对象的话,需要序列化和然后转化为二进制数组。
                    for (int i = 0; i < 100; i++)
                    {
                        body = Encoding.UTF8.GetBytes("这是第-----"+i+"-----条消息");
                        //channel.BasicPublish("", "firstQueue", null, body);
                        channel.BasicPublish("NewExchange", "newRoutingKey", properties, body); 
                        Console.Write("成功发送第-----"+i+"-----条消息!");
                    }
                    Console.ReadKey();
                }
            }
复制代码

 

消费者a:

 

 

复制代码
static void Main(string[] args)
        {
            var factory = new ConnectionFactory();
            factory.HostName = "127.0.0.1";
            factory.UserName = "feiyang";
            factory.Password = "123456";
            using (var connection  = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    //创建一个新的,持久的交换区  
                    channel.ExchangeDeclare("NewExchange", ExchangeType.Direct, true, false, null);  
                    //还是连接到哪个队列
                    channel.QueueDeclare("newQueue",true,false,false,null);
                    // 绑定队列到交换区  
                    channel.QueueBind("newQueue", "NewExchange", "newRoutingKey"); 
                    //定义消息接受者
                    var customer = new QueueingBasicConsumer(channel);
                    //从指定队列获取消息,
                    //中间这个参数实际必须打开,为false,意思是是否不通知rabbitm已经处理完毕,我们这里要设成false,要通知
                    //channel.BasicConsume("firstQueue",true,customer);
                    channel.BasicConsume("newQueue", false, customer);
                    //由于队列分发不公平导致一个压力很大,一个很小,在这设置下,公平q分发,也就是一个消费者处理完通知队列后,才会继续分发一个
                    channel.BasicQos(0, 1, false);
                    //开始不断循环出队列的消息
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)customer.Queue.Dequeue();
                        //将消息二进制转回字符串
                        var msg = Encoding.UTF8.GetString(ea.Body);
                        //通知队列,已经处理完成
                        channel.BasicAck(ea.DeliveryTag, false);
                        Console.WriteLine(msg);
                    }
                    //sw.Stop();
                    //Console.WriteLine("共用时" + sw.ElapsedTicks + "毫秒");
                    //Console.ReadKey();
                }
            }
        }
复制代码

消费者b:

复制代码
 var factory = new ConnectionFactory();
            factory.HostName = "127.0.0.1";
            factory.UserName = "feiyang";
            factory.Password = "123456";
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    //创建一个新的,持久的交换区  
                    //channel.ExchangeDeclare("NewExchange", ExchangeType.Direct, true, false, null);
                    //还是连接到哪个队列
                    channel.QueueDeclare("newQueue", true, false, false, null);
                    // 绑定队列到交换区  
                    //channel.QueueBind("newQueue", "NewExchange", "newRoutingKey");
                    //定义消息接受者
                    var customer = new QueueingBasicConsumer(channel);
                    //从指定队列获取消息,
                    //中间这个参数实际必须打开,为false,意思是是否不通知rabbitm已经处理完毕,我们这里要设成false,要通知
                    //channel.BasicConsume("firstQueue",true,customer);
                    channel.BasicConsume("newQueue", false, customer);
                    //由于队列分发不公平导致一个压力很大,一个很小,在这设置下,公平q分发,也就是一个消费者处理完通知队列后,才会继续分发一个
                    channel.BasicQos(0, 1, false);
                    //开始不断循环出队列的消息
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)customer.Queue.Dequeue();
                        //将消息二进制转回字符串
                        var msg = Encoding.UTF8.GetString(ea.Body);
                        //通知队列,已经处理完成
                        channel.BasicAck(ea.DeliveryTag, false);
                        Console.WriteLine(msg);
                    }
                    //sw.Stop();
                    //Console.WriteLine("共用时" + sw.ElapsedTicks + "毫秒");
                    //Console.ReadKey();
                }
            }
复制代码

运行结果

 

2.1个生产者,2个消费者,路由类型direct,匹配规则routingKey相同,匹配不同的队列,一次发送到2个队列各个消费者取出各自的队列消息。

生产者,创建一个交换区,创建一个队列,

复制代码
 //创建一个队列
                    //队列是否持久化
                    bool durable = true;
                    //已经存在的队列,不能再定义持久化
                   // channel.QueueDeclare("firstQueue",false,false,false,null);
                    //创建一个新的,持久的交换区  
                    channel.ExchangeDeclare("queueExchange", ExchangeType.Direct, true, false, null);  
                    //持久的队列, 没有排他性,与不自动删除 
                    channel.QueueDeclare("queue_a", durable, false, false, null);
                    // 绑定队列到交换区  
                    channel.QueueBind("queue_a", "queueExchange", "queueRoutingKey"); 
                    //消息持久化
                    var properties = channel.CreateBasicProperties();
                    properties.SetPersistent(true);
                    properties.DeliveryMode = 2; //消息是持久的,存在并不会受服务器重启影响  
                    byte[] body = null;
                    //消息是以二进制数组的形式传输的,所以如果消息是实体对象的话,需要序列化和然后转化为二进制数组。
                    for (int i = 0; i < 100; i++)
                    {
                        body = Encoding.UTF8.GetBytes("这是第-----"+i+"-----条消息");
                        //channel.BasicPublish("", "firstQueue", null, body);
                        channel.BasicPublish("queueExchange", "queueRoutingKey", properties, body); 
                        Console.Write("成功发送第-----"+i+"-----条消息!");
                    }
复制代码

消费者a,创建一个新队列,绑定到和生产者同一个交换区,读取刚刚创建的新队列数据。

复制代码
 //创建一个新的,持久的交换区  
                    //channel.ExchangeDeclare("NewExchange", ExchangeType.Direct, true, false, null);  
                    //还是连接到哪个队列
                    channel.QueueDeclare("queue_a_b", true, false, false, null);
                    // 绑定队列到交换区  
                    channel.QueueBind("queue_a_b", "queueExchange", "queueRoutingKey"); 
                    //定义消息接受者
                    var customer = new QueueingBasicConsumer(channel);
                    //从指定队列获取消息,
                    //中间这个参数实际必须打开,为false,意思是是否不通知rabbitm已经处理完毕,我们这里要设成false,要通知
                    //channel.BasicConsume("firstQueue",true,customer);
                    channel.BasicConsume("queue_a_b", false, customer);
                    //由于队列分发不公平导致一个压力很大,一个很小,在这设置下,公平q分发,也就是一个消费者处理完通知队列后,才会继续分发一个
                    channel.BasicQos(0, 1, false);
                    //开始不断循环出队列的消息
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)customer.Queue.Dequeue();
                        //将消息二进制转回字符串
                        var msg = Encoding.UTF8.GetString(ea.Body);
                        //通知队列,已经处理完成
                        channel.BasicAck(ea.DeliveryTag, false);
                        Console.WriteLine(msg);
                    }
复制代码

消费b,直接读取生产者创建的queue_a队列消息

复制代码
 //创建一个新的,持久的交换区  
                    //channel.ExchangeDeclare("NewExchange", ExchangeType.Direct, true, false, null);
                    //还是连接到哪个队列
                    //channel.QueueDeclare("newQueue", true, false, false, null);
                    // 绑定队列到交换区  
                    //channel.QueueBind("newQueue", "NewExchange", "newRoutingKey");
                    //定义消息接受者
                    var customer = new QueueingBasicConsumer(channel);
                    //从指定队列获取消息,
                    //中间这个参数实际必须打开,为false,意思是是否不通知rabbitm已经处理完毕,我们这里要设成false,要通知
                    //channel.BasicConsume("firstQueue",true,customer);
                    channel.BasicConsume("queue_a", false, customer);
                    //由于队列分发不公平导致一个压力很大,一个很小,在这设置下,公平q分发,也就是一个消费者处理完通知队列后,才会继续分发一个
                    channel.BasicQos(0, 1, false);
                    //开始不断循环出队列的消息
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)customer.Queue.Dequeue();
                        //将消息二进制转回字符串
                        var msg = Encoding.UTF8.GetString(ea.Body);
                        //通知队列,已经处理完成
                        channel.BasicAck(ea.DeliveryTag, false);
                        Console.WriteLine(msg);
                    }
复制代码

运行结果

可以看见,通过路由匹配,一次发送消息,发送到匹配到的两个队列中,两个消费者各自读取各自的队列。

3.篇幅有限,再来一个路由类型为Topic的代码例子。

生产者,由于已经创建了一个queueexChange类型为direct的交换区,不能更改类型,所以重新创建一个交换区

复制代码
  //创建一个队列
                    //队列是否持久化
                    bool durable = true;
                    //已经存在的队列,不能再定义持久化
                   // channel.QueueDeclare("firstQueue",false,false,false,null);
                    //创建一个新的,持久的交换区  
                    channel.ExchangeDeclare("queueTopicExchange", ExchangeType.Topic, true, false, null);  
                    //持久的队列, 没有排他性,与不自动删除 
                    channel.QueueDeclare("queue.a", durable, false, false, null);
                    // 绑定队列到交换区  
                    channel.QueueBind("queue.a", "queueTopicExchange", "queue.#"); 
                    //消息持久化
                    var properties = channel.CreateBasicProperties();
                    properties.SetPersistent(true);
                    properties.DeliveryMode = 2; //消息是持久的,存在并不会受服务器重启影响  
                    byte[] body = null;
                    //消息是以二进制数组的形式传输的,所以如果消息是实体对象的话,需要序列化和然后转化为二进制数组。
                    for (int i = 0; i < 100; i++)
                    {
                        body = Encoding.UTF8.GetBytes("这是第-----"+i+"-----条消息");
                        //channel.BasicPublish("", "firstQueue", null, body);
                        channel.BasicPublish("queueTopicExchange", "queue.#", properties, body); 
                        Console.Write("成功发送第-----"+i+"-----条消息!");
                    }
复制代码

消费者a:

复制代码
 //创建一个新的,持久的交换区  
                    //channel.ExchangeDeclare("NewExchange", ExchangeType.Direct, true, false, null);  
                    //还是连接到哪个队列
                    channel.QueueDeclare("queue.a.b", true, false, false, null);
                    // 绑定队列到交换区  
                    channel.QueueBind("queue.a.b", "queueTopicExchange", "queue.#"); 
                    //定义消息接受者
                    var customer = new QueueingBasicConsumer(channel);
                    //从指定队列获取消息,
                    //中间这个参数实际必须打开,为false,意思是是否不通知rabbitm已经处理完毕,我们这里要设成false,要通知
                    //channel.BasicConsume("firstQueue",true,customer);
                    channel.BasicConsume("queue.a.b", false, customer);
                    //由于队列分发不公平导致一个压力很大,一个很小,在这设置下,公平q分发,也就是一个消费者处理完通知队列后,才会继续分发一个
                    channel.BasicQos(0, 1, false);
                    //开始不断循环出队列的消息
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)customer.Queue.Dequeue();
                        //将消息二进制转回字符串
                        var msg = Encoding.UTF8.GetString(ea.Body);
                        //通知队列,已经处理完成
                        channel.BasicAck(ea.DeliveryTag, false);
                        Console.WriteLine(msg);
                    }
复制代码

消费者b:

复制代码
 //创建一个新的,持久的交换区  
                    //channel.ExchangeDeclare("NewExchange", ExchangeType.Direct, true, false, null);
                    //还是连接到哪个队列
                    //channel.QueueDeclare("newQueue", true, false, false, null);
                    // 绑定队列到交换区  
                    //channel.QueueBind("newQueue", "NewExchange", "newRoutingKey");
                    //定义消息接受者
                    var customer = new QueueingBasicConsumer(channel);
                    //从指定队列获取消息,
                    //中间这个参数实际必须打开,为false,意思是是否不通知rabbitm已经处理完毕,我们这里要设成false,要通知
                    //channel.BasicConsume("firstQueue",true,customer);
                    channel.BasicConsume("queue.a", false, customer);
                    //由于队列分发不公平导致一个压力很大,一个很小,在这设置下,公平q分发,也就是一个消费者处理完通知队列后,才会继续分发一个
                    channel.BasicQos(0, 1, false);
                    //开始不断循环出队列的消息
                    while (true)
                    {
                        var ea = (BasicDeliverEventArgs)customer.Queue.Dequeue();
                        //将消息二进制转回字符串
                        var msg = Encoding.UTF8.GetString(ea.Body);
                        //通知队列,已经处理完成
                        channel.BasicAck(ea.DeliveryTag, false);
                        Console.WriteLine(msg);
                    }
复制代码

运行结果:

代码例子就不一一写出来了,还有很多种情况,实际项目根据不同的需求灵活运用,有兴趣的可以自己搭配测试一下。

    本站是提供个人知识管理的网络存储空间,所有内容均由用户发布,不代表本站观点。如发现有害或侵权内容,请点击这里 或 拨打24小时举报电话:4000070609 与我们联系。

    猜你喜欢

    0条评论

    发表

    请遵守用户 评论公约

    类似文章 更多
    喜欢该文的人也喜欢 更多