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

news2024/11/6 7:25:04

介绍

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();
            }
        }
    }