.NET Core使用RabbitMQ

RabbitMQ简介

　RabbitMQ是一个开源的,基于AMQP(Advanced Message Queuing Protocol)协议的完整的可复用的企业级消息队,RabbitMQ可以实现点对点,发布订阅等消息处理模式。

RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写，支持Linux,windows,macOS,FreeBSD等操作系统,同时也支持很多语言,如：Python,Java,Ruby,PHP,C#,JavaScript,Go,Elixir,Objective-C,Swift等。

RabbitMQ安装

我使用的环境是ubuntu18.04,

RabbitMq需要Erlang语言的支持，在安装RabbitMq之前需要安装Erlang

sudo apt-get install erlang-nox
更新源

sudo apt-get update
安装RabbitMq

sudo apt-get install rabbitmq-server
添加users用户，密码设置为admin

sudo rabbitmqctl add_user users admin
给添加的用户赋予权限

sudo rabbitmqctl set_user_tags users administrator
赋予virtual host中所有资源的配置、写、读权限以便管理其中的资源

rabbitmqctl set_permissions -p users '.*' '.*' '.*'
官方提供的一个web管理工具(rabbitmq_management),定位到Rabbitmq安装目录然后启动web控制台

sudo rabbitmq-plugins enable rabbitmq_management
成功后可以在浏览器输入http://localhost:15672/查看RabbitMq信息

RabbitMQ常用命令

启动

sudo rabbitmq-server start
停止

sudo rabbitmq-server stop
重启

sudo rabbitmq-server restart
查看状态

sudo rabbitmqctl status
查看所有用户

rabbitmqctl list_users
查看用户权限

rabbitmqctl list_user_permissions users
删除用户权限

rabbitmqctl clear_permissions [-p VHostPath] users
删除用户

rabbitmqctl delete_user users
修改用户密码

rabbitmqctl change_password users newpassword

.NET Core 使用RabbitMQ

通过install-package rabbitmq.client命令或nuget安装rabbitmq.client包

生产者实现

using System;

using System.Text;

using RabbitMQ.Client;

namespace RabbitMQ

{

    class MainClass

    {

        static void Main(string[] args)

        {

            Console.WriteLine("生产者");

            IConnectionFactory factory = new ConnectionFactory//创建连接工厂对象

            {

                HostName = "106.12.90.208",//IP地址

                Port = 5672,//端口号

                UserName = "admin",//用户账号

                Password = "admin"//用户密码

            };

            IConnection con = factory.CreateConnection();//创建连接对象

            IModel channel = con.CreateModel();//创建连接会话对象

            string name = "demo";

            //声明一个队列

            channel.QueueDeclare(

              queue: name,//消息队列名称

              durable: false,//是否持久化,true持久化,队列会保存磁盘,服务器重启时可以保证不丢失相关信息。

              exclusive: false,//是否排他,true排他的,如果一个队列声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除.

              autoDelete: false,//是否自动删除。true是自动删除。自动删除的前提是：致少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时,才会自动删除.

              arguments: null //设置队列的一些其它参数

               );

            string str = string.Empty;

            do

            {

                Console.WriteLine("发送内容:");

                str = Console.ReadLine();

                //消息内容

                byte[] body = Encoding.UTF8.GetBytes(str);

                //发送消息

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

                Console.WriteLine("成功发送消息:" + str);

            }while(str.Trim().ToLower() != "exit");

            con.Close();

            channel.Close();

        }

    }

}

消费者实现

using System;

using System.Text;

using RabbitMQ.Client;

using RabbitMQ.Client.Events;

namespace RabbitMQ

{

    class MainClass

    {

        static void Main(string[] args)

        {

            Console.WriteLine("消费者");

            IConnectionFactory factory = new ConnectionFactory//创建连接工厂对象

            {

                HostName = "106.12.90.208",//IP地址

                Port = 5672,//端口号

                UserName = "admin",//用户账号

                Password = "admin"//用户密码

            };

            IConnection conn = factory.CreateConnection();

            IModel channel = conn.CreateModel();

            string name = "demo";

            //声明一个队列

            channel.QueueDeclare(

              queue: name,//消息队列名称

              durable: false,//是否持久化,true持久化,队列会保存磁盘,服务器重启时可以保证不丢失相关信息。

              exclusive: false,//是否排他,true排他的,如果一个队列声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除.

              autoDelete: false,//是否自动删除。true是自动删除。自动删除的前提是：致少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时,才会自动删除.

              arguments: null ////设置队列的一些其它参数

               );

            //创建消费者对象

            var consumer = new EventingBasicConsumer(channel);

            consumer.Received += (model, ea) =>

            {

                byte[] message = ea.Body;//接收到的消息

                        Console.WriteLine("接收到消息为:" + Encoding.UTF8.GetString(message));

            };

            //消费者开启监听

            channel.BasicConsume(name, true, consumer);

            Console.ReadKey();

            channel.Dispose();

            conn.Close();

        }

    }

}

同时运行两个项目看效果

RabbitMQ的Worker模式

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

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

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

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

consumer.Received += (model, ea) =>

{

    Thread.Sleep(3000);

    byte[] message = ea.Body;

    Console.WriteLine("接收到消息为:" + Encoding.UTF8.GetString(message));

};

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

RabbitMQ消息确认

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

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

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

consumer.Received += (model, ea) =>

{

    byte[] message = ea.Body;

    Console.WriteLine("接收到消息为:" + Encoding.UTF8.GetString(message));

    Thread.Sleep(3000); //等待三秒手动确认

    channel.BasicAck(ea.DeliveryTag, true);//返回消息确认

};

////将autoAck设置false 关闭自动确认.

channel.BasicConsume(name, false, consumer);

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

‘能者多劳’模式

能者多劳是给消费速度快的消费更多的消息.少的责消费少的消息.能者多劳是建立在手动确认基础上实现。

在延迟确认的消费中添加BasicQos

channel.QueueDeclare(

    queue: name,//消息队列名称

    durable: false,//是否持久化,true持久化,队列会保存磁盘,服务器重启时可以保证不丢失相关信息。

    exclusive: false,//是否排他,true排他的,如果一个队列声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除.

    autoDelete: false,//是否自动删除。true是自动删除。自动删除的前提是：致少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时,才会自动删除.

    arguments: null ////设置队列的一些其它参数

    );

//每次只能向消费者发送一条信息,再消费者未确认之前,不再向他发送信息

channel.BasicQos(0,1,false);

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

Exchange模式

在RabbitMQ的Exchange模式中生产者并不会直接把消息发送到Queue中,而是将消息发送到Exchange（交换机),消费者创建各自的队列绑定到交换机.

发布订阅模式(fanout)

生产者实现, 把队列替换成了交换机, 发布消息时把交换机名称告诉RabbitMQ,把交换机设置成fanout发布订阅模式

using System;

using System.Text;

using RabbitMQ.Client;

namespace RabbitMQ

{

    class MainClass

    {

        static void Main(string[] args)

        {

            Console.WriteLine("生产者");

            IConnectionFactory factory = new ConnectionFactory//创建连接工厂对象

            {

                HostName = "106.12.90.208",//IP地址

                Port = 5672,//端口号

                UserName = "admin",//用户账号

                Password = "admin"//用户密码

            };

            IConnection con = factory.CreateConnection();//创建连接对象

            IModel channel = con.CreateModel();//创建连接会话对象

            sstring exchangeName = "exchange1";//交换机名称

            //把交换机设置成fanout发布订阅模式

            channel.ExchangeDeclare(name, type: "fanout");

            string str;

            do

            {

                str = Console.ReadLine();

                //消息内容

                byte[] body = Encoding.UTF8.GetBytes(str);

                //发布消息,

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

            }while(str.Trim().ToLower() != "exit");

            con.Close();

            channel.Close();

        }

    }

}

消费者实现

using System;

using System.Text;

using System.Threading;

using RabbitMQ.Client;

using RabbitMQ.Client.Events;

namespace mq

{

    class MainClass

    {

        static void Main(string[] args)

        {

            IConnectionFactory factory = new ConnectionFactory//创建连接工厂对象

            {

                HostName = "106.12.90.208",//IP地址

                Port = 5672,//端口号

                UserName = "admin",//用户账号

                Password = "admin"//用户密码

            };

            IConnection conn = factory.CreateConnection();

            IModel channel = conn.CreateModel();

            //交换机名称

            string exchangeName = "exchange1";

            //声明交换机

            channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout);

            //消息队列名称

            string queueName = DateTime.Now.Second.ToString();

            //声明队列

            channel.QueueDeclare(queueName, false, false, false, null);

            //将队列与交换机进行绑定

            channel.QueueBind(queueName, exchangeName, "", null);

            //定义消费者

            var consumer = new EventingBasicConsumer(channel);

            Console.WriteLine($"队列名称:{queueName}");

            //接收事件

            consumer.Received += (model, ea) =>

            {

                byte[] message = ea.Body;//接收到的消息

                Console.WriteLine($"接收到信息为:{Encoding.UTF8.GetString(message)}");

                //返回消息确认

                channel.BasicAck(ea.DeliveryTag, true);

            };

            //开启监听

            channel.BasicConsume(queueName, false, consumer);

            Console.ReadKey();

        }

    }

}

当消费者绑定同样的交换机,可以看到两个不同的消费者都能接受到生产者发送的所有消息。

路由模式(Direct Exchange)

路由模式下,在发布消息时指定不同的routeKey,交换机会根据不同的routeKey分发消息到不同的队列中

生产者实现

using System;

using System.Text;

using RabbitMQ.Client;

namespace RabbitMQ

{

    class MainClass

    {

        static void Main(string[] args)

        {

            Console.WriteLine("生产者");

            IConnectionFactory factory = new ConnectionFactory//创建连接工厂对象

            {

                HostName = "106.12.90.208",//IP地址

                Port = 5672,//端口号

                UserName = "admin",//用户账号

                Password = "admin"//用户密码

            };

            IConnection con = factory.CreateConnection();//创建连接对象

            IModel channel = con.CreateModel();//创建连接会话对象

            string exchangeName = "exchange1"; //交换机名称

            string routeKey = "key1"; //匹配的key，

            //把交换机设置成Direct模式

            channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);

            string str;

            do

            {

                str = Console.ReadLine();

                //消息内容

                byte[] body = Encoding.UTF8.GetBytes(str);

                //发送消息

                channel.BasicPublish(exchangeName, routeKey, null, body);

            }while(str.Trim().ToLower() != "exit");

            con.Close();

            channel.Close();

        }

    }

}

申明一个routeKey值为key1,并在发布消息的时候告诉了RabbitMQ,消息传递时routeKey必须匹配,才会被队列接收否则消息会被抛弃。

消费者实现

using System;

using System.Text;

using RabbitMQ.Client;

using RabbitMQ.Client.Events;

namespace mq

{

    class MainClass

    {

        static void Main(string[] args)

        {

            Console.WriteLine($"输入接受key名称:");

            string routeKey = Console.ReadLine();

            IConnectionFactory factory = new ConnectionFactory//创建连接工厂对象

            {

                HostName = "106.12.90.208",//IP地址

                Port = 5672,//端口号

                UserName = "admin",//用户账号

                Password = "admin"//用户密码

            };

            IConnection conn = factory.CreateConnection();

            IModel channel = conn.CreateModel();

            //交换机名称

            string exchangeName = "exchange11";

            //声明交换机

            channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);

            //消息队列名称

            string queueName = DateTime.Now.Second.ToString();

            //声明队列

            channel.QueueDeclare(queueName, false, false, false, null);

            //将队列,key与交换机进行绑定

            channel.QueueBind(queueName, exchangeName, routeKey, null);

            //定义消费者

            var consumer = new EventingBasicConsumer(channel);

            Console.WriteLine($"队列名称:{queueName}");

            //接收事件

            consumer.Received += (model, ea) =>

            {

                byte[] message = ea.Body;//接收到的消息

                Console.WriteLine($"接收到信息为:{Encoding.UTF8.GetString(message)}");

                //返回消息确认

                channel.BasicAck(ea.DeliveryTag, true);

            };

            //开启监听

            channel.BasicConsume(queueName, false, consumer);

            Console.ReadKey();

        }

    }

}

在routeKey匹配的时候才会接收消息,接收者消息队列可以声明多个routeKey与交换机进行绑定

routeKey不匹配则不接收消息。

通配符模式(Topic Exchange)

通配符模式和路由模式其实差不多,不同于配符模式中的路由可以声明为模糊查询.

符号“#”匹配一个或多个词.

符号“*”匹配一个词。

RabbitMQ中通配符的通配符是用"."来分割字符串的.比如a.*只能匹配到a.b,a.c,而a.#可以匹配到a.a.c,a.a.b.

生成者实现

using System;

using System.Text;

using RabbitMQ.Client;

namespace RabbitMQ

{

    class MainClass

    {

        static void Main(string[] args)

        {

            Console.WriteLine("生产者");

            IConnectionFactory factory = new ConnectionFactory//创建连接工厂对象

            {

                HostName = "106.12.90.208",//IP地址

                Port = 5672,//端口号

                UserName = "admin",//用户账号

                Password = "admin"//用户密码

            };

            IConnection con = factory.CreateConnection();//创建连接对象

            IModel channel = con.CreateModel();//创建连接会话对象

            string exchangeName = "exchange114"; //交换机名称

            string routeKey = "key.a"; //匹配的key，

            //把交换机设置成Topic模式

            channel.ExchangeDeclare(exchangeName, ExchangeType.Topic);

            string str;

            do

            {

                str = Console.ReadLine();

                //消息内容

                byte[] body = Encoding.UTF8.GetBytes(str);

                //发送消息

                channel.BasicPublish(exchangeName, routeKey, null, body);

            }while(str.Trim().ToLower() != "exit");

            con.Close();

            channel.Close();

        }

    }

}

消费者实现

using System;

using System.Text;

using RabbitMQ.Client;

using RabbitMQ.Client.Events;

namespace mq

{

    class MainClass

    {

        static void Main(string[] args)

        {

            Console.WriteLine($"输入接受key名称:");

            string routeKey = "key.*"; //使用通配符来匹配key

            IConnectionFactory factory = new ConnectionFactory//创建连接工厂对象

            {

                HostName = "106.12.90.208",//IP地址

                Port = 5672,//端口号

                UserName = "admin",//用户账号

                Password = "admin"//用户密码

            };

            IConnection conn = factory.CreateConnection();

            IModel channel = conn.CreateModel();

            //交换机名称

            string exchangeName = "exchange114";

            //声明交换机

            channel.ExchangeDeclare(exchangeName, ExchangeType.Topic);

            //消息队列名称

            string queueName = DateTime.Now.Second.ToString();

            //声明队列

            channel.QueueDeclare(queueName, false, false, false, null);

            //将队列与交换机进行绑定

            channel.QueueBind(queueName, exchangeName, routeKey, null);

            //定义消费者

            var consumer = new EventingBasicConsumer(channel);

            Console.WriteLine($"队列名称:{queueName}");

            //接收事件

            consumer.Received += (model, ea) =>

            {

                byte[] message = ea.Body;//接收到的消息

                Console.WriteLine($"接收到信息为:{Encoding.UTF8.GetString(message)}");

                //返回消息确认

                channel.BasicAck(ea.DeliveryTag, true);

            };

            //开启监听

            channel.BasicConsume(queueName, false, consumer);

            Console.ReadKey();

        }

    }

}

只有在通配符匹配通过的情况下才会接收消息,

巴特西

.NET Core使用RabbitMQ

RabbitMQ简介

RabbitMQ安装

RabbitMQ常用命令

.NET Core 使用RabbitMQ

RabbitMQ的Worker模式

RabbitMQ消息确认

‘能者多劳’模式

Exchange模式

发布订阅模式(fanout)

路由模式(Direct Exchange)

通配符模式(Topic Exchange)

最新文章

热门文章