C#使用RabbitMQ-1_Docker部署并在c#中实现简单模式消息代理

本是古典何须时尚 2024-04-23 20:23 47阅读 0赞

## 介绍 ##

**RabbitMQ是一个开源的消息队列系统，实现了高级消息队列协议（AMQP）**。

?RabbitMQ起源于金融系统，现在广泛应用于各种分布式系统中。它的主要功能是在应用程序之间提供异步消息传递，实现系统间的解耦和消息的可靠传递。RabbitMQ使用Erlang语言开发，支持多种客户端语言如Python、Ruby、.NET、Java等。

此外，RabbitMQ具有以下特点：

1.  **易用性**：提供了简单易用的API，使得生产者和消费者可以方便地发送和接收消息。
2.  **扩展性**：可以水平扩展以处理大量的消息，支持集群部署来提高系统的吞吐量和可用性。
3.  **高可用性**：通过镜像队列等机制保证消息不会因服务器故障而丢失，确保了系统的稳健运行。
4.  **多种交换模式**：支持直接交换、扇形交换、主题交换和头交换等多种交换模式，满足不同的消息路由需求。
5.  **多协议支持**：除了AMQP协议，还支持STOMP等其他消息协议。

## 在docker中部署RabbitMQ ##

?首先在dockerHub中找到RabbitMQ的镜像

[rabbitmq - 官方图片 (docker.com)][rabbitmq - _ _docker.com]

?执行命令拉取镜像

docker pull rabbitmq

?镜像拉取完成之后启动镜像

docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq

?此时我们打开 [http://localhost:15672/][http_localhost_15672]会发现无法访问，这是因为管理插件还未被激活

![9852509663084be280ee9e6ec8d2d699.png][]

?通过`docker ps -a`查看部署的RabbitMQ容器id，在通过使用命令

`docker exec -it 容器id /bin/bash`

?进入容器内部再运行：`rabbitmq-plugins enable rabbitmq_management`

![7d8af826e3cd46b6a99b4310ffcab788.png][]

?此时就可以打开管理插件了，第一次使用 RabbitMQ 管理界面，需要使用默认的用户名和密码（ `guest`/`guest`）来登录

![053b1b3f0d094d698bf4c6eaee28a738.png][]

## 消息队列简单模式（直连交换机） ##

?simple模式，是RabbitMQ最简单的一种模式，如下图所示，只有一个生产者，一个消费者和一个队列

![3b28f322150d43ef81be0bb4ead18030.png][]

?生产者和消费者在发送和接受消息时，只需要指定队列名，而不需要指定发送到哪个Exchange

### 生产者代码 ###

class MyClass
    {
        public static void Main(string[] args)
        {
            var factory = new ConnectionFactory();
            factory.HostName = "localhost"; //RabbitMQ服务在本地运行
            factory.UserName = "guest"; //用户名
            factory.Password = "guest"; //密码
    
            //创建连接
            using (var connection = factory.CreateConnection())
            {
                //创建通道
                using (var channel = connection.CreateModel())
                {
                    //声明一个名称为hello的消息队列
                    channel.QueueDeclare("hello", false, false, false, null); 
    
                    for (int i = 0; i < 5; i++)
                    {
                        string message = "Hello Word ! " + i; //传递的消息内容
                        var body = Encoding.UTF8.GetBytes(message);
                        //此处的参数"hello" 就对应的就是上面声明的消息队列的路由键
                        channel.BasicPublish("", "hello", null, body); //开始传递
                        Console.WriteLine("已发送： {0}", message);
                    }
                }
            }
        }
    }

### 消费者代码（自动模式） ###

class MyClass
    {
        static void Main(string[] args)
        {
            //创建连接工厂
            var factory = new ConnectionFactory();
            factory.HostName = "localhost";
            factory.UserName = "guest";
            factory.Password = "guest";
            //创建连接
            using (var connection = factory.CreateConnection())
            {
                //创建通道
                using (var channel = connection.CreateModel())
                {
                    //声明队列
                    channel.QueueDeclare("hello", false, false, false, null);
                    
                    //事件的基本消费者
                    var consumer = new EventingBasicConsumer(channel);
                    
                    consumer.Received += (model, ea) =>
                    {
                        var body = ea.Body.ToArray();
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine("已接收： {0}", message);
                        //发送消息确认信号（手动确认）    
                        //channel.BasicAck(ea.DeliveryTag,false);
                    };
                    
                    //当 autoAck设置为true时，也就是自动确认模式，一旦消息队列将消息发送给消息消费者后，就会从内存中将这个消息删除。
                    //当autoAck设置为false时，也就是手动模式，如果此时的有一个消费者宕机，消息队列就会将这条消息继续发送给其他的消费者，这样数据在消息消费者集群的环境下，就不会不丢失了。
                    channel.BasicConsume("hello", true, consumer);
                    Console.ReadKey();
                }
            }
        }
    }

?在消费者代码中，要注意的是autoAck设置为true时，也就是自动确认模式的时候，要去掉手动发送确认信号代码：channel.BasicAck(ea.DeliveryTag,false);

### 代码调试 ###

![b4d0076abd3b4f3895f833cc0c371722.png][]

? 在代码执行到

channel.QueueDeclare("hello", false, false, false, null);

?声明一个消息队列时，在RabbitMQ的可视化界面就可以看见多了一个名称为hello的消息队列

![ad1f4292b9aa49a5913782cb296b8fdb.png][]

?当执行完五次消息传递时，点击上图的hello消息队列，Messages:设置为5，然后点击Get Messages，就可以看到传递过来的5条消息队列了

![70395a463a71417fa70cc4e295974ece.png][]

![06e380bc4cdd4614b25f97ddecd8ec1e.png][]

?此时我们执行消费者代码，因为autoAck设置为true了，当消息队列将消息发送给消费者后，就会立马将消息删除，此时再点击Get Messages就会提示Queue is empty

![7a3bceb813dc47d4bca1d68742f038cf.png][]

### 消费者代码（手动模式） ###

在手动模式中，我们将原来声明的队列删除调，然后将生产者与消费者的声明队列代码中的第二个参数都改为true，意味着这个队列是持久的。在 RabbitMQ 服务器重启之后，持久的队列和它的消息都不会丢失

channel.QueueDeclare("hello", true, false, false, null);

然后添加一行代码，设置当前消费者的预取模式为只预取一条消息

channel.BasicQos(0, 1, false);

接着是 发送消息确认信号 与 开启手动确认模式

channel.BasicAck(ea.DeliveryTag,false);
    channel.BasicConsume("hello", false, consumer);

要注意的地方是，一定要在结尾处加上下方代码，否则可能会出现还没发送消息确认信号，进程就结束了，这时就会发现将接收到的helloword打印到控制台了，去RabbitMQ可视乎管理界面发现事件还未被消耗掉。

Console.ReadKey();

static void Main(string[] args)
        {
            //创建连接工厂
            var factory = new ConnectionFactory();
            factory.HostName = "localhost";
            factory.UserName = "guest";
            factory.Password = "guest";
            //创建连接
            using (var connection = factory.CreateConnection())
            {
                //创建通道
                using (var channel = connection.CreateModel())
                {
                    //声明队列
                    channel.QueueDeclare("hello", true, false, false, null);
                    channel.BasicQos(0, 1, false);
                    //事件的基本消费者
                    var consumer = new EventingBasicConsumer(channel);
                    
                    consumer.Received += (model, ea) =>
                    {
                        var body = ea.Body.ToArray();
                        var message = Encoding.UTF8.GetString(body);
                        Console.WriteLine("已接收： {0}", message);
                        //发送消息确认信号（手动确认）    
                        channel.BasicAck(ea.DeliveryTag,false);
                    };
                    
                    //当 autoAck设置为true时，也就是自动确认模式，一旦消息队列将消息发送给消息消费者后，就会从内存中将这个消息删除。
                    //当autoAck设置为false时，也就是手动模式，如果此时的有一个消费者宕机，消息队列就会将这条消息继续发送给其他的消费者，这样数据在消息消费者集群的环境下，就不会不丢失了。
                    channel.BasicConsume("hello", false, consumer);
                    Console.ReadKey();
                }
            }
        }

### 代码调试 ###

按照上面改成手动模式后，消费者只预取一条消息，如下图0、1、2被消费掉了，此时再去RabbitMQ的可视乎管理界面点击GetMessage，只剩下3、4的消息内容了

![4aec2ff58b7444198bff08fcff118070.png][]

![e2bffaae82fd48eb8f8b870edcde6fd5.png][]

[rabbitmq - _ _docker.com]: https://hub.docker.com/_/rabbitmq
[http_localhost_15672]: http://localhost:15672/
[9852509663084be280ee9e6ec8d2d699.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/a4a9d52b06eb4ff6a0f8efaf02b10515.png
[7d8af826e3cd46b6a99b4310ffcab788.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/1df4972594b44a4aa1bee84e9118c9c4.png
[053b1b3f0d094d698bf4c6eaee28a738.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/54def1bda9e94e90b0168bbc56974519.png
[3b28f322150d43ef81be0bb4ead18030.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/3ea014bdb70e42778860ea4b6bcd51cb.png
[b4d0076abd3b4f3895f833cc0c371722.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/5f0a6b7c2dc4445baaff81249cb5eb36.png
[ad1f4292b9aa49a5913782cb296b8fdb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/9e39d58fd6b24727ac84b789b835f2fe.png
[70395a463a71417fa70cc4e295974ece.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/8a995dcc00e44f0783b1ab143bb1d4fc.png
[06e380bc4cdd4614b25f97ddecd8ec1e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/9e3ff55420994776baff781f6c78b80b.png
[7a3bceb813dc47d4bca1d68742f038cf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/a072408d17a245e3b1d2a9f627708c11.png
[4aec2ff58b7444198bff08fcff118070.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/834ac6a9f5474703b6d5d42e2d8404ed.png
[e2bffaae82fd48eb8f8b870edcde6fd5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/3c48a067914146c09d7fb3f74f2b88c0.png