.NET Core RabbitMQ探索(1)

忘是亡心i 2021-12-09 15:53 269阅读 0赞

RabbitMQ可以被比作一个邮局，当你向邮局寄一封信时，邮局会保证将这封信送达你写的收件人，而RabbitMQ与邮局最主要的区别是，RabbitMQ并不真的处理信件，它处理的是二进制的数据块，它除了接收“信件”，他还要负责处理数据(比如保存数据)、转发数据(实际上送信是邮递员的职责)等。

在RabbitMQ中有以下几个概念：

*  **生产者**

一个生产者仅仅负责向RabbitMQ中发送消息，除此之外它不会完成其他工作。

*  **队列**

队列由RabbitMQ自身提供，它的作用就像是邮箱和邮局的关系一样。虽然看似RabbitMQ和你的应用直接通信，但是实际上消息是被存储到队列中的，实际上，队列就是一个消息缓冲区，受宿主环境内存和磁盘空间的限制。多个生产者可以将消息发送到一个队列，而多个消费者可以从一个队列中接收消息。

*  **消费者**

一个消费者仅仅负责接收来自生产者的消息。

这里需要注意以下两点：

1.  生产者、消费者和RabbitMQ不一定必须在相同的宿主机上，在实际的生产环境中，它们常常处于不同的宿主机；
2.  某个应用既可以同时是一个生产者和消费者。

我们需要建立两个.net core项目，分别是发布端、接收端，它们由消息队列进行连接，消息队列会保存消息，直到消息被发送给接收端，如图所示。

![1225995-20190517213905993-233952982.jpg][]

下面开始创建项目：

![1225995-20190517214003755-1963596306.png][]

创建项目后，两个项目均需安装RabbitMQ.Client的Nuget包，并restore。

![1225995-20190517214112596-1771935364.png][]

在正式开始编写发送端代码之前，我们需要安装RabbitMQ，这里我在Docker上安装RabbitMQ，所用系统为CentOS7。  
　　首先创新新的文件夹目录/docker/rabbitmq/data，用作RabbitMQ容器的数据卷挂载。

![1225995-20190517214217456-1943522917.png][]

然后运行下面的语句，如果本地没有镜像，Docker会自动去Dockerhub上拉取镜像：

docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq --volume /rabbitmq/data:/var/lib/rabbitmq rabbitmq:management

镜像下载完毕后可以看到容器已经运行，其中，15671用于外网访问，5672用于内网连接。

![1225995-20190517214405231-444717342.jpg][]

访问http://ip:15671，不出意外你可以进入到RabbitMQ的可视化管理界面。

![1225995-20190517214448673-147119408.jpg][]

*  生产者端

下面可以开始编写生产者端的代码，如下所示。

using System;
    using System.Text;
    using RabbitMQ.Client;
    
    namespace Send
    {
        class Program
        {
            static void Main(string[] args)
            {
                var factory = new ConnectionFactory() { HostName = "148.70.210.208" };
                //连接RabbitMQ
                using (var connection = factory.CreateConnection())
                {
                    //创建一个信道
                    using (var channel = connection.CreateModel())
                    {
                        //声明一个Hello队列，信道就与Hello队列连接起来了
                        channel.QueueDeclare(queue: "hello",
                            durable: false,
                            exclusive: false,
                            autoDelete: false,
                            arguments: null);
    
                        string message = "Hello World!";
                        //消息体
                        var body = Encoding.UTF8.GetBytes(message);
    
                        //通过信道，将消息发送到队列
                        //需要指明队列的名称，才能转发到特定队列
                        channel.BasicPublish(exchange: "",
                            routingKey: "hello",
                            basicProperties: null,
                            body: body);
    
                        Console.WriteLine(" [x] Sent [0]", message);
                    }
    
                    Console.WriteLine("Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }
    }

程序首先通过Socket与RabbitMQ进行了连接，连接成功后，创建了一个信道，并通过这个信道声明了一个名叫"hello"的队列，这样，队列就和这个信道绑定起来，在发送数据时，只需要在routingKey中指明所用的队列名，消息就会通过direct exchange的模式被发送到指明名称的队列中，这就好像你在网上买了商品，然后你告诉卖家，一定要发某个快递一样，这样，你的商品就会通过这个快递公司邮寄给你。

direct exchange模式如图所示。

![1225995-20190517214737052-1089630367.png][]

整个流程可以如下进行概括，首先建立发布端和RabbitMQ进行Socket连接，并且声明Socket连接中的信道，通过这个信道声明特定名称的队列，这样，数据就可以通过这个信道发送到RabbitMQ的队列中，等待被转发到消费端，这与官网给出的流程图如出一辙。

![1225995-20190517214818843-65861055.png][]

*  消费者端

下面我们就可以开始编写消费端的程序，对于消费者端来说，它一直在进行监听，我们需要一直运行我们的监听程序，一旦收到消息，就把消息通过控制台进行打印。  
　　打开我们的Recieve项目，代码如下所示。

using System;
    using System.Text;
    using RabbitMQ.Client;
    using RabbitMQ.Client.Events;
    
    namespace Receive
    {
        class Program
        {
            static void Main(string[] args)
            {
                var factory = new ConnectionFactory() { HostName = "your ip address" };
                using (var connection = factory.CreateConnection())
                {
                    using (var channel = connection.CreateModel())
                    {
                        channel.QueueDeclare(queue: "hello",
                            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);
                             Console.WriteLine($"Received message：{message}");
                         };
    
                        channel.BasicConsume(queue: "hello",
                        autoAck: true,
                        consumer: consumer);
    
                        Console.WriteLine("Press [enter] to exit");
                        Console.ReadLine();
                    }
                }
            }
        }
    }

一开始的代码与发送端一致，建立连接、创建信道、声明队列，需要注意这里声明的队列名需要与发送端绑定的队列名一致。这里之所以重新声明队列，是为了保证当接收端程序比发送端先启动时，队列依然是存在的，但是需要保证名称一样。

以上准备工作完成后，接收端可以开始异步的接收队列中的数据了。我们使用EventBasicConsumer.Received事件来绑定接收完成的回调函数。即：当消费端成功从"hello"队列中接收到数据后，就会调用这个匿名的lambda表达式，从消息体中获取数据，将数据转码成字符串格式并且打印到控制台中，至此，一条消息的处理就完毕了。  
　　需要注意，在消费端收到信息后，需要给客户端确认，即返回autoAck，如果没有返回，客户端会默认消费者没有收到，消息会存储在队列中，直到被消费。这里设置autoAck:true，则表示消费者一旦收到消息，则立刻自动的向客户端回应，这样，消息就会从队列中删除。

*  测试代码

运行代码可以看到，无论先运行消费端还是生产者端的代码，消费端都能收到消息：

![1225995-20190517215026588-1477508463.png][]

如果先运行生产者端程序，消息会存储在消息队列中，当运行消费者端程序后，保存在消息队列中的消息将会被消费，进而从队列中删除。  
　　例如，我们先不运行消费者端程序，只运行生产者端，这是我们刷新RabbitMQ的可视化界面。

![1225995-20190517215114317-1149026192.jpg][]

可以看到有一条消息处于Ready状态，但是还未被消费，所以被存储，我们关闭生产者端程序，再次运行，可以看到队列中未被消费的消息变为两条。

![1225995-20190517215157165-1983496311.jpg][]

这时，我们运行消费者端程序，运行成功后可以看到，消费者端一次收到了两条消息，并且队列中的消息被清除。

![1225995-20190517215234150-2111873844.jpg][]

![1225995-20190517215256766-1981399893.jpg][]

由此可见，对于RabbitMQ的消息队列，生产者和消费者端是完全分离的。  
　　下一讲我们将继续探索RabbitMQ的工作队列。

转载于:https://www.cnblogs.com/cquptjyz/p/10883882.html

[1225995-20190517213905993-233952982.jpg]: /images/20211209/e310cce4241f40b184028d262c09dcae.png
[1225995-20190517214003755-1963596306.png]: /images/20211209/53485c31a9f04b30876858777761def6.png
[1225995-20190517214112596-1771935364.png]: /images/20211209/a69fc83406fe4be6adf0b7d10a520cd4.png
[1225995-20190517214217456-1943522917.png]: /images/20211209/ebc7e288d00e47c78ca2d38c45c4c964.png
[1225995-20190517214405231-444717342.jpg]: /images/20211209/972c2bb2b9d44144a7f45875b594ff69.png
[1225995-20190517214448673-147119408.jpg]: /images/20211209/f3fafd0a6ccf493cad4b88a2f333bd76.png
[1225995-20190517214737052-1089630367.png]: /images/20211209/101f6b303275425b8757a778329393ec.png
[1225995-20190517214818843-65861055.png]: /images/20211209/fcc83b675ec6484fb33dc6fdac5292df.png
[1225995-20190517215026588-1477508463.png]: /images/20211209/c5903491cb3348aa94fc76bcc36549ed.png
[1225995-20190517215114317-1149026192.jpg]: /images/20211209/e81623ae93d647649cc1b735551a2978.png
[1225995-20190517215157165-1983496311.jpg]: /images/20211209/5765ba23c7774aaa85d0dc6018b0be21.png
[1225995-20190517215234150-2111873844.jpg]: /images/20211209/6f9a1da54dd3469db29aa18e9f737f11.png
[1225995-20190517215256766-1981399893.jpg]: /images/20211209/c47c4156ca22416fa056a91231a70db9.png