.Net下RabbitMQ的使用(2) -- 发送接收消息

在安装和配置完成RabbitMQ之后，我们就可以尝试做一个最简单的例子：发送和接收消息。

 

我们先来看客户端也就是发送者的代码：

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Xml.Serialization;

namespace SendService
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory();
            factory.HostName = "192.168.12.111";
            factory.Port = 5672;
            factory.UserName = "admin";
            factory.Password = "admin";

            //定义要发送的数据

            List<RequestMessage> messages = new List<RequestMessage>();
            for (int i = 0; i < 200;i++ )
            {
                RequestMessage message = new RequestMessage() { MessageId = Guid.NewGuid(), Message = "this is a 请求。" + i };
                messages.Add(message);
            }
            
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("all.sms.message", false, false, false, null);//hello是queue的名字
                    //序列化消息对象，RabbitMQ并不支持复杂对象的序列化，所以对于自定义的类型需要自己序列化
                    foreach (var item in messages)
                    {
                        XmlSerializer xs = new XmlSerializer(typeof(RequestMessage));
                        using (MemoryStream ms = new MemoryStream())
                        {

                            xs.Serialize(ms, item);

                            byte[] bytes = ms.ToArray();

                            //指定发送的路由，通过默认的exchange直接发送到指定的队列中。

                            channel.BasicPublish("", "all.sms.message", null, bytes);
                            Console.WriteLine(" [x] Sent {0}", item.Message);
                        }
                    }
                    //var body = Encoding.UTF8.GetBytes(message);
                    //channel.BasicPublish("", "all.sms.message", null, body);//hello是routing key的名字
                     
                    Console.ReadLine();
                }
            }
        }
    }

    public class RequestMessage
    {
        public Guid MessageId { set; get; }
        public string Message { set; get; }
    }
}

 

在方法

channel.BasicPublish("", "esbtest.rmq.consoleserver", null, bytes);

中的第一个参数是需要输入一个exchange。在RabbitMQ中，所有的消息都必须要通过exchange发送到各个queue里面去。发送者发送消息，其实也就是把消息放到exchange中去。而exchange知道应该把消息放到哪里去。在这个方法中，我们没有输入exchange的名称，只是定义了一个空的echange，而在第二个参数routeKey中输入了我们目标队列的名称。RabbitMQ会帮我定义一个默认的exchange，这个exchange会把消息直接投递到

我们输入的队列中，这样服务端只需要直接去这个定义了的队列中获取消息就可以了。

服务端的代码：

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Xml.Serialization;

namespace ReceiveService
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory();
            factory.HostName = "192.168.12.111";
            factory.Port = 5672;
            factory.UserName = "admin";
            factory.Password = "admin";
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("all.sms.message", false, false, false, null);//hello是queue的名字
                    Console.WriteLine("Listening...");
                    //在队列上定义一个消费者
                    var consumer = new QueueingBasicConsumer(channel);
                    channel.BasicConsume("all.sms.message", true, consumer);//hello是queue的名字,这里可以理解为hello是routing key的名字。因为这个例子没有使用指定名称的exchange（实际上使用的是默认的exchange名字），所以queue的名字和routing key的名字是相同的。在第五篇文章中介绍如果使用了指定名称的exchange，queue name和routing key的关系与用法。
                    Console.WriteLine(" [*] Waiting for messages." + "To exit press CTRL+C");
                    while (true)
                    {
                        //阻塞函数，获取队列中的消息
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//阻塞
                        byte[] body = ea.Body;
                        XmlSerializer xs = new XmlSerializer(typeof(RequestMessage));
                        using (MemoryStream ms = new MemoryStream(body))
                        {

                            RequestMessage message = (RequestMessage)xs.Deserialize(ms);

                            Console.WriteLine("Receive a Message, Id:" + message.MessageId + " Message:" + message.Message);

                        }
                        //Console.ReadLine();
                        //var message = Encoding.UTF8.GetString(body);
                        //Console.WriteLine(" [x] Received {0}", message);
                    }
                }
            }  
        }
    }

    public class RequestMessage
    {
        public Guid MessageId { set; get; }
        public string Message { set; get; }
    }
}

 

至此，简单的发送接收程序就可以运行了，运行RabbitMQ，然后分别运行客户端和服务端。运行结果如图：

客户端：

 

服务端：