当前位置：首页 > news >正文

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

news 2025/6/30 10:44:03

介绍

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

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

此外，RabbitMQ具有以下特点：

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

在docker中部署RabbitMQ

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

rabbitmq - 官方图片 (docker.com)

🍀执行命令拉取镜像

docker pull rabbitmq

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

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

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

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

docker exec -it 容器id /bin/bash

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

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

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

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

🍀生产者和消费者在发送和接受消息时，只需要指定队列名，而不需要指定发送到哪个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);

代码调试

🍀 在代码执行到

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

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

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

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

消费者代码（手动模式）

在手动模式中，我们将原来声明的队列删除调，然后将生产者与消费者的声明队列代码中的第二个参数都改为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();}}}