RabbitMQ原理与相关操作(一)

我不是女神ヾ 2022-06-06 12:55 206阅读 0赞

# [RabbitMQ原理与相关操作(一)][RabbitMQ] #

小编是菜鸟一枚，最近想试试MQ相关的技术，所以自己看了下RabbitMQ官网，试着写下自己的理解与操作的过程。

刚开始的第一篇，原理只介绍 生产者、消费者、队列，至于其他的内容，会在后续中陆续补齐。

# 引入MQ话题 #

## 什么时候会用到MQ ##

可能很多人有疑惑：MQ到底是什么？哪些场景下要使用MQ？  
前段时间安装了RabbitMQ，现在就记录下自己的学习心得吧。  
首先看段程序：

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

class Program
        {
            static void Main(string[] args)
            {
                new Thread(Write).Start();
                new Thread(Write).Start();
                new Thread(Write).Start();
                new Thread(Write).Start();
            }
            
            public static void WriteLog(int i)
            {
                using (FileStream f = new FileStream(@"d:\\test.txt", FileMode.Append))
                {
                    using (StreamWriter sw = new StreamWriter(f, Encoding.Default))
                    {
                        sw.Write(i);
                    }
                }
            }
    
            public static void Write()
            {
                for (int i = 0; i < 10000; i++)
                {
                    WriteLog(i);
                }
            }
        }

![复制代码][copycode.gif]

仅仅从代码上看，没有觉得任何问题对吧？编译也是通过的，但是执行时，出现一个问题：

![784082-20160928165046297-270372959.png][]

当然，这仅仅是一个小的案例，类似这种多线程写文件造成的问题， 就应该使用MQ了。

MQ的使用场景大概包括解耦，提高峰值处理能力，送达和排序保证，缓冲等。

## MQ概述 ##

消息队列技术是分布式应用间交换信息的一种技术。

消息队列可驻留在内存或磁盘上,队列存储消息直到它们被应用程序读走。

通过消息队列，应用程序可独立地执行--它们不需要知道彼此的位置、或在继续执行前不需要等待接收程序接收此消息。

MQ主要作用是接受和转发消息。你可以想想在生活中的一种场景：当你把信件的投进邮筒，邮递员肯定最终会将信件送给收件人。我们可以把MQ比作 邮局和邮递员。

MQ和邮局的主要区别是,它不处理消息,但是,它会接受数据、存储消息数据、转发消息。

## RabbitMQ术语 ##

生产者：

消息发送者，在MQ中被称为生产者（producer），一个发送消息的应用也被叫做生产者，用P表示

![784082-20160928231957938-1176094746.png][]

### 消费者： ###

生产者“生产”出消息后，最终由谁消费呢？等待接受消息的应用程序，我们称之为消费者（Consuming ），用C表示

![784082-20160928233231719-482776638.png][]

### 队列： ###

消息只能存储在队列（queue ）中。尽管消息在rabbitMQ和应用程序间流通，但是队列却是存在于RabbitMQ内部。

一个队列不受任何限制，它可以存储你想要存储的消息量，它本质上是一个无限的缓冲区。

多个生产者可以向同一个队列发送消息，多个消费者可以尝试从同一个消息队列中接收数据。

一个队列像下面这样（上面是它的队列名称）

![784082-20160928232904094-1668513314.png][]

### 注意： ###

生产者、消费者、中间件不必在一台机器上，实际应用中也是绝大多数不在一起的。我们可以用一张图表示RabbitMQ的构造：

![784082-20160928233939516-93411931.png][]

注：此图片摘自于[百度百科RabbitMQ][RabbitMQ 1]。

# 使用RabbitMQ解决多线程写入文件问题 #

## 分析 ##

多线程写入，产生消息的也就是一个程序（一个生产者P），消费消息的也是一个消息，它的模型应该是：

![784082-20160928235701469-244168332.png][]

## 编写代码 ##

### 引入RabbitMQ client DLL ###

程序包管理控制台命令：

PM> Install-Package RabbitMQ.Client

![784082-20160929000034797-1825496409.png][]

![784082-20160929000223219-2039278212.png][]

### 生产者 ###

首先，创建一个 connection 通过socket连接 去和服务器连接起来（需要传目的服务器的IP、用户名、密码等）。

接着 创建一个 channel ，这是大部分的要做的事情所在。

要发送消息，我们必须声明一个队列，，然后我们可以向队列发布消息。

执行一次BasicPublish方法，推送一个消息。

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

class Program
        {
            static void Main(string[] args)
            {
                new Thread(Write).Start();
                new Thread(Write).Start();
                new Thread(Write).Start();
            }
    
            public static void Write()
            {
                var factory = new ConnectionFactory() { HostName = "localhost", UserName = "eric", Password = "123456", };
                using (var connection = factory.CreateConnection())
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare(queue: "writeLog", durable: false, exclusive: false, autoDelete: false, arguments: null);
                    for (int i = 0; i < 8000; i++)
                    {
                        string message = i.ToString();
                        var body = Encoding.UTF8.GetBytes(message);
    
                        channel.BasicPublish(exchange: "", routingKey: "writeLog", basicProperties: null, body: body);
                        Console.WriteLine("Program Sent {0}", message);
                    }
                }
            }
        }

![复制代码][copycode.gif]

声明的队列，在服务器中如果不存在了，会自动创建。而消息的内容是字节数组，在使用时，注意编码问题。

### 消费者 ###

当队列里有消息时，消费者要随时能够从队列里获取消息，所以我需要一直运行它，让它监听消息。

就像我们打篮球进行传球，需要事先确认要传给的那个队友位置一样，生产者要发送消息，一定要事先知道消费消息的程序的对列是哪个。所以，在运行生产者程序前，需要先启动消费者程序。

由此，声明对列，就应该在消费者程序中完成。

![ExpandedBlockStart.gif][]

![复制代码][copycode.gif]

class Program
        {
            public static void Main()
            {
                var factory = new ConnectionFactory() { HostName = "localhost", UserName = "eric", Password = "123456", VirtualHost ="/"};
                using (var connection = factory.CreateConnection())
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare(queue: "writeLog",
                                         durable: false,
                                         exclusive: false,
                                         autoDelete: false,
                                         arguments: null);
    
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);
                        ExcuateWriteFile(message);
                        Console.WriteLine(" Receiver Received {0}", message);
                    };
                    channel.BasicConsume(queue: "writeLog",
                                         noAck: true,
                                         consumer: consumer);
    
                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
            public static void ExcuateWriteFile(string i)
            {
                using (FileStream fs = new FileStream(@"d:\\test.txt", FileMode.Append))
                {
                    using (StreamWriter sw = new StreamWriter(fs, Encoding.Unicode))
                    {
                        sw.Write(i);
                    }
                }
            }
        }

![复制代码][copycode.gif]

# 执行程序 #

先执行 消费者程序，让它一直保持监听。

## 错误解决 ##

执行时VS报错：

“RabbitMQ.Client.Exceptions.BrokerUnreachableException”类型的未经处理的异常在 RabbitMQ.Client.dll 中发生 其他信息: None of the specified endpoints were reachable。

进入查看详细的内部异常：

innerEception:\{"The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=530, text=\\"NOT\_ALLOWED - access to vhost '/' refused for user 'eric'\\", classId=10, methodId=40, cause="\}

此时，我们打开在[http://localhost:15672/\#/users][http_localhost_15672_users] 可以看到eric 下 的Can access virtual hosts 为 NoAccess

![784082-20160929013756360-1848893754.png][]

解决办法：

rabbitmqctl控制台输入

rabbitmqctl set_permissions -p / userName "." "." ".*"

![784082-20160929014034078-1624241672.png][]

![784082-20160929014222438-254413496.png][]

再次执行时，可以看到：

![784082-20160929015524703-1596114417.png][]

然后运行 生产者程序。

我们先开着 Receive ，当生产者运行时

![784082-20160929014913844-970735671.png][]

消费者的自动触发执行 ：

![784082-20160929014948953-520805058.png][]

直到所有的 指定的 queue 里面的消息完全消费完为止。（此时消费者程序仍然在监听中）

对于需要安装和设置用户的同学，请参考 [windows下 安装 rabbitMQ 及操作常用命令][windows_ _ rabbitMQ]

本文参考：

[rabbitMq外文网][rabbitMq]站

[百度百科][Link 1]

作者：[Eric Li][]   
出处：[http://www.cnblogs.com/ericli-ericli/][Eric Li]   
除转载文章外，随笔版权归作者和博客园所有，欢迎转载，转载请标明出处。   
如果您觉得本篇博文对您有所收获，觉得作者还算用心，请点击右下角的 \[推荐\]，谢谢！

标签:  [RabbitMQ 消息队列][RabbitMQ 2]

[RabbitMQ]: http://www.cnblogs.com/ericli-ericli/p/5917018.html
[ExpandedBlockStart.gif]: /images/20220606/2c7c133e4d4742ce8f060c6eaf93408b.png
[copycode.gif]: /images/20220606/abbf3d2a53d8488ab53c210d38abc992.png
[784082-20160928165046297-270372959.png]: /images/20220606/beb90f313a6547449a89567a47d3352d.png
[784082-20160928231957938-1176094746.png]: /images/20220606/70d86fa186dd4cb9bf81135dc19b4946.png
[784082-20160928233231719-482776638.png]: /images/20220606/4b19fc4f763544579c83841e2a82b641.png
[784082-20160928232904094-1668513314.png]: /images/20220606/4bc58403cc4c4064b93ac6410aeed4dc.png
[784082-20160928233939516-93411931.png]: /images/20220606/ee599965f25e425c89e6448aaba1cab6.png
[RabbitMQ 1]: http://baike.baidu.com/link?url=GoSlvtJi6zkEpbGkvd5XzOGauMGdzN7vRe2PfpvSwefup5cx79DNi46EdgaIJ8209WhaGWUlfx8Vs-PqBNm1iq
[784082-20160928235701469-244168332.png]: /images/20220606/32c7e43c02904c56b6690e481cb1380a.png
[784082-20160929000034797-1825496409.png]: /images/20220606/3bea18d068fc474cb2764b89bf6ab4f6.png
[784082-20160929000223219-2039278212.png]: /images/20220606/255afe9749014ffab9b0fb0ce82a9a9f.png
[http_localhost_15672_users]: http://localhost:15672/#/users
[784082-20160929013756360-1848893754.png]: /images/20220606/ec4ca1ed13bc47b18890eac1f9a10c0d.png
[784082-20160929014034078-1624241672.png]: /images/20220606/e7fbee22452d4e0685220d7fe69ab927.png
[784082-20160929014222438-254413496.png]: /images/20220606/0bff13252fa44fd2a23d9220ca57252c.png
[784082-20160929015524703-1596114417.png]: /images/20220606/1d2b97f50dce4c88b23510943d1d918c.png
[784082-20160929014913844-970735671.png]: /images/20220606/65c3d05c524847759e2c3cabb25329c7.png
[784082-20160929014948953-520805058.png]: /images/20220606/9876126f7b5a4f859ee708b7b74d4833.png
[windows_ _ rabbitMQ]: http://www.cnblogs.com/ericli-ericli/p/5902270.html
[rabbitMq]: http://www.rabbitmq.com/tutorials/tutorial-one-dotnet.html
[Link 1]: http://baike.baidu.com/link?url=na27yrlNr2_Ti2Utcnr3KYVxywHwqCzgw3hXuZWRiN1MEKZkPN_nqjAb_1P0MA90hfI0FnO7BVr1FYb7RI6N3a
[Eric Li]: http://www.cnblogs.com/ericli-ericli/
[RabbitMQ 2]: http://www.cnblogs.com/ericli-ericli/tag/RabbitMQ%20%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97/