[c#]RabbitMQ的简单使用

2022-10-01 22:35:14

摘要

Message Queue消息队列，简称MQ，是一种应用程序对应用程序的通信方法，应用程序通过读写出入队列的消息来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此通信。

MQ是消费-生产者模型的一个典型代表，一端往消息队列中不断写入消息，而另一端则可以读取或者订阅队列中的消息。

安装

官网地址：http://www.rabbitmq.com/download.html。

下载后，需要安装RabbitMQ服务。如果没有Erlang环境会弹出下面的提示：

下载Erlang环境并安装

地址：http://www.erlang.org/downloads

然后安装RabbitMQ，安装成功后会在服务中看到该服务。

一个例子

新建一个控制台应用程序，并使用Nuget引入RabbitMQ

代码如下：

当type等于1时，客户端为生产者，2时为消费者

    /// <summary>

    /// RabbitMQ

    /// </summary>

    class Program

    {

        static void Main(string[] args)

        {

            string type = Console.ReadLine();

            //生产者

            if (type=="")

            {

                ConnectionFactory factory = new ConnectionFactory();

                factory.HostName = "127.0.0.1";

                //默认端口

                factory.Port = ;

                using (IConnection conn = factory.CreateConnection())

                {

                    using (IModel channel = conn.CreateModel())

                    {

                        //在MQ上定义一个持久化队列，如果名称相同不会重复创建

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

                        while (true)

                        {

                            string message = string.Format("Message_{0}", Console.ReadLine());

                            byte[] buffer = Encoding.UTF8.GetBytes(message);

                            IBasicProperties properties = channel.CreateBasicProperties();

                            properties.DeliveryMode = ;

                            channel.BasicPublish("", "MyRabbitMQ", properties, buffer);

                            Console.WriteLine("消息发送成功：" + message);

                        }

                    }

                }

            }

            else

            {

                //消费者

                ConnectionFactory factory = new ConnectionFactory();

                factory.HostName = "127.0.0.1";

                //默认端口

                factory.Port = ;

                using (IConnection conn = factory.CreateConnection())

                {

                    using (IModel channel = conn.CreateModel())

                    {

                        //在MQ上定义一个持久化队列，如果名称相同不会重复创建

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

                        //输入1，那如果接收一个消息，但是没有应答，则客户端不会收到下一个消息

                        channel.BasicQos(, , false);

                        Console.WriteLine("Listening...");

                        //在队列上定义一个消费者

                        QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);

                        //消费队列，并设置应答模式为程序主动应答

                        channel.BasicConsume("MyRabbitMQ", false, consumer);

                        while (true)

                        {

                            //阻塞函数，获取队列中的消息

                            BasicDeliverEventArgs ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                            byte[] bytes = ea.Body;

                            string str = Encoding.UTF8.GetString(bytes);

                            Console.WriteLine("队列消息:" + str.ToString());

                            //回复确认

                            channel.BasicAck(ea.DeliveryTag, false);

                        }

                    }

                }

            }

        }

    }