欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

.NET Core使用RabbitMQ

程序员文章站 2022-04-05 18:06:45
rabbitmq简介 rabbitmq是一个开源的,基于amqp(advanced message queuing protocol)协议的完整的可复用的企业级消息队,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信息
    .NET Core使用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();
        }

    }
}

同时运行两个项目看效果
.NET Core使用RabbitMQ

rabbitmq的worker模式

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

.NET Core使用RabbitMQ

默认情况下,rabbitmq会顺序的将message发给下一个消费者。每个消费者会得到平均数量的message。这种方式称之为round-robin(轮询).
但是很多情况下并不希望消息平均分配,而是要消费快的多消费,消费少的少消费.还有很多情况下一旦其中一个宕机,那么另外接收者的无法接收原本这个接收者所要接收的数据。
我们修改其中一个消费者代码,让其等待3秒。在等待中停止运行 看看效果

consumer.received += (model, ea) =>
{
    thread.sleep(3000);
    byte[] message = ea.body;
    console.writeline("接收到消息为:" + encoding.utf8.getstring(message));
};

.NET Core使用RabbitMQ

当消费者宕机后消费者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);

.NET Core使用RabbitMQ

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

‘能者多劳’模式

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

channel.queuedeclare(
    queue: name,//消息队列名称
    durable: false,//是否持久化,true持久化,队列会保存磁盘,服务器重启时可以保证不丢失相关信息。
    exclusive: false,//是否排他,true排他的,如果一个队列声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除.
    autodelete: false,//是否自动删除。true是自动删除。自动删除的前提是:致少有一个消费者连接到这个队列,之后所有与这个队列连接的消费者都断开时,才会自动删除.
    arguments: null ////设置队列的一些其它参数
    );
//每次只能向消费者发送一条信息,再消费者未确认之前,不再向他发送信息
channel.basicqos(0,1,false);

.NET Core使用RabbitMQ

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

exchange模式

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

发布订阅模式(fanout)

.NET Core使用RabbitMQ

生产者实现, 把队列替换成了交换机, 发布消息时把交换机名称告诉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();
        }
    }
}

.NET Core使用RabbitMQ

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

路由模式(direct exchange)

.NET Core使用RabbitMQ

路由模式下,在发布消息时指定不同的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();
        }
    }

}

.NET Core使用RabbitMQ

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

.NET Core使用RabbitMQ

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