RabbitMQ 消息队列

Posted 逍遥帝君

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ 消息队列相关的知识,希望对你有一定的参考价值。

一:简介

  RabbitMQ是一个在AMQP协议标准基础上完整的,可服用的企业消息系统。他遵循Mozilla Public License开源协议。采用 Erlang 实现的工业级的消息队列(MQ)服务器

RabbitMQ的官方站:http://www.rabbitmq.com/ 
     AMQP(高级消息队列协议) 是一个异步消息传递所使用的应用层协议规范,作为线路层协议,而不是API(例如JMS),AMQP 客户端能够无视消息的来源任意发送和接受信息。AMQP的原始用途只是为金融界提供一个可以彼此协作的消息协议,而现在的目标则是为通用消息队列架构提供通用构建工具。因此,面向消息的中间件 (MOM)系统,例如发布/订阅队列,没有作为基本元素实现。反而通过发送简化的AMQ实体,用户被赋予了构建例如这些实体的能力。这些实体也是规范的一 部分,形成了在线路层协议顶端的一个层级:AMQP模型。这个模型统一了消息模式,诸如之前提到的发布/订阅,队列,事务以及流数据,并且添加了额外的特性,例如更易于扩展,基于内容的路由。 

 

1.下载Erlang

http://www.erlang.org/download.html

有32位与64位版本可供选择,根据个人需要下载相应版本,下载好后先安装Erlang

2.下载RabbitMQ服务器安装文件

 http://www.rabbitmq.com/install-windows.html

安装好RabbitMQ服务器端环境后,RabbitMQ将会以服务的形式驻留在服务器上,默认是开启状态

.在 cmd 中指向 sbin 目录,并输入以下命令

rabbitmq-plugins enable rabbitmq_management

http://localhost:15672,在登陆界面输入用户名:guest,密码:guest,即可进入管理界面查看各种信息 

二:RabbitMQ 示例

发送消息端

  class Program
    {
        static void Main(string[] args)
        {
            UserHelp mes =new UserHelp();
            while (true)
            {
                var result = mes.GetUserMessage();
                Console.WriteLine( result);
            }

        }
    }

 

 public string GetUserMessage()
        {
            try
            {
                var conFactory = new ConnectionFactory();
                conFactory.HostName = "localhost";
                conFactory.UserName = "wangdongsheng";
                conFactory.Password = "123456";
                conFactory.VirtualHost = "/";

                using (var connect = conFactory.CreateConnection())
                {
                    //rabbitmq服务端
                    using (var channel = connect.CreateModel())
                    {
                        //在MQ上定义一个持久化队列,如果名称相同不会重复创建
                        channel.QueueDeclare("TestQueue", true, false, false, null);
                        // //输入1,那如果接收一个消息,但是没有应答,则客户端不会收到下一个消息
                        channel.BasicQos(0, 1, false);
                        //在队列上定义一个消费者
                        var consumer = new QueueingBasicConsumer(channel);
                        //消费队列,并设置应答模式为程序主动应答
                        channel.BasicConsume("TestQueue", false, consumer);
                        while (true)
                        {
                            //阻塞函数,获取队列中的消息
                            var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
                            byte[] bytes = ea.Body;
                            string str = Encoding.UTF8.GetString(bytes);
                            var msg = JsonConvert.DeserializeObject<RequestMsg>(str);
                            //恢复确认
                            channel.BasicAck(ea.DeliveryTag, false);
                            return "接收的数据:"+"发送人-"+msg.Name+";发送内容-"+msg.Code;
                        }

                    }
                }
            }
            catch (Exception ex)
            {
                return ex.Message;
            }
        }
   public class RequestMsg
    {
        public string Name { get; set; }
        public string Code { get; set; }
    }

接收消息端

 class Program
    {
        private static void Main(string[] args)
        {

            SendHelp help=new SendHelp();
            while (true)
            {
                Console.WriteLine("请输入内容:");
                var write = Console.ReadLine();
                if (!string.IsNullOrWhiteSpace(write))
                {
                    var result = help.SendMessage(write);
                    Console.ForegroundColor = ConsoleColor.DarkYellow;
                    Console.WriteLine(result);
                    Console.ForegroundColor = ConsoleColor.White;
                }
            }
            Console.ReadKey();
        }
    }
        public string SendMessage(string msg)
        {
            try
            {
                var conFactory = new ConnectionFactory();
                conFactory.HostName = "localhost";
                conFactory.UserName = "wangdongsheng";
                conFactory.Password = "123456";
                conFactory.VirtualHost = "/";

                using (IConnection connect = conFactory.CreateConnection())
                {
                    //rabbitmq服务端
                    using (var channel = connect.CreateModel())
                    {
                        //在MQ上定义一个持久化队列,如果名称相同不会重复创建
                        channel.QueueDeclare("TestQueue", true, false, false, null);
                        while (true)
                        {
                            var requetMsg = new RequestMsg();
                            requetMsg.Name = "东升";
                            requetMsg.Code = msg;
                            string jsonStr = JsonConvert.SerializeObject(requetMsg);
                            byte[] bytes = Encoding.UTF8.GetBytes(jsonStr);
                            IBasicProperties pro = channel.CreateBasicProperties();
                            pro.DeliveryMode = 2;
                            channel.BasicPublish("", "TestQueue", pro, bytes);
                            return "发送消息成功:发送人-"+requetMsg.Name+";消息内容-" + requetMsg.Code;
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                return "发送的消息为:" + ex.Message;
            }
        }

 

1.Server(broker): 接受客户端连接,实现AMQP消息队列和路由功能的进程。

2.Virtual Host:其实是一个虚拟概念,类似于权限控制组,一个Virtual Host里面可以有若干个Exchange和Queue,但是权限控制的最小粒度是Virtual Host

3.Exchange:接受生产者发送的消息,并根据Binding规则将消息路由给服务器中的队列。ExchangeType决定了Exchange路由消息的行为,例如,在RabbitMQ中,ExchangeType有direct、Fanout和Topic三种,不同类型的Exchange路由的行为是不一样的。

4.Message Queue:消息队列,用于存储还未被消费者消费的消息。

5.Message: 由Header和Body组成,Header是由生产者添加的各种属性的集合,包括Message是否被持久化、由哪个Message Queue接受、优先级是多少等。而Body是真正需要传输的APP数据。

6.Binding:Binding联系了Exchange与Message Queue。Exchange在与多个Message Queue发生Binding后会生成一张路由表,路由表中存储着Message Queue所需消息的限制条件即Binding Key。当Exchange收到Message时会解析其Header得到Routing Key,Exchange根据Routing Key与Exchange Type将Message路由到Message Queue。Binding Key由Consumer在Binding Exchange与Message Queue时指定,而Routing Key由Producer发送Message时指定,两者的匹配方式由Exchange Type决定。 

7.Connection:连接,对于RabbitMQ而言,其实就是一个位于客户端和Broker之间的TCP连接。

8.Channel:信道,仅仅创建了客户端到Broker之间的连接后,客户端还是不能发送消息的。需要为每一个Connection创建Channel,AMQP协议规定只有通过Channel才能执行AMQP的命令。一个Connection可以包含多个Channel。之所以需要Channel,是因为TCP连接的建立和释放都是十分昂贵的,如果一个客户端每一个线程都需要与Broker交互,如果每一个线程都建立一个TCP连接,暂且不考虑TCP连接是否浪费,就算操作系统也无法承受每秒建立如此多的TCP连接。RabbitMQ建议客户端线程之间不要共用Channel,至少要保证共用Channel的线程发送消息必须是串行的,但是建议尽量共用Connection。

9.Command:AMQP的命令,客户端通过Command完成与AMQP服务器的交互来实现自身的逻辑。例如在RabbitMQ中,客户端可以通过publish命令发送消息,txSelect开启一个事务,txCommit提交一个事务。

 

以上是关于RabbitMQ 消息队列的主要内容,如果未能解决你的问题,请参考以下文章

RabbitMQ学习笔记五:RabbitMQ之优先级消息队列

RabbitMQ:第二章:Spring整合RabbitMQ(简单模式,广播模式,路由模式,通配符模式,消息可靠性投递,防止消息丢失,TTL,死信队列,延迟队列,消息积压,消息幂等性)(代码

Rabbitmq 消息队列

微服务专题之.Net6下集成消息队列-RabbitMQ交换机模式代码演示(全)

RabbitMQ 消息队列学习

RabbitMQ确认机制问题处理