RabbitMq的简单使用

Posted 2021-11-29 wikiz

本篇将介绍RabbitMq的一个简单使用例子，分别介绍生产者如何发送消息，消费者如何接收和处理消息

关于RabbitMQ的知识背景的文章非常多。我对它的总结是，解决高并发请求的瓶颈，将应用程序真正处理请求的担子推送给一个处理消息的中心，以减轻应用程序本身处理请求的压力。

这个处理中心也即RabbitMQ的消费者，同时，它跟生产者一样，也是RabbitMQ的客户端。

下面将使用RabbitMQ.Client来创建一个生产者客户端，一个消费者客户端：

生产者代码：

        static void Main(string[] args)
        {
            var facotry = new ConnectionFactory()
            {
                HostName = "192.168.1.23",
                VirtualHost = "/host",
                UserName = "admin",
                Password = "123456"
            };

            using (var con = facotry.CreateConnection())
            {
                //3. 创建信道
                using (var channel = con.CreateModel())
                {
                    //4. 申明队列
                    channel.QueueDeclare(queue: "wiky_test_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);

                    //5. 构建byte消息数据包
                    string message = args.Length > 0 ? args[0] : "Hello RabbitMQ!";

                    var body = Encoding.UTF8.GetBytes(message);

                    //6. 发送数据包
                    channel.BasicPublish(exchange: "", routingKey: "wiky_test_queue", basicProperties: null, body: body);

                    Console.WriteLine(" [x] Sent {0}", message);
                }
            }

        }

消费者代码：

        static void Main(string[] args)
        {

            //1.实例化连接工厂
            var facotry = new ConnectionFactory()
            {
                HostName = "192.168.1.23",
                VirtualHost = "/host",
                UserName = "admin",
                Password = "123456"
            };
            //2. 建立连接
            using (var connection = facotry.CreateConnection())
            {
                //3. 创建信道
                using (var channel = connection.CreateModel())
                {
                    //4. 申明队列
                    channel.QueueDeclare(queue: "wiky_test_queue", durable: false, exclusive: false, autoDelete: false, arguments: null);
                    //5. 构造消费者实例
                    var consumer = new EventingBasicConsumer(channel);
                    //6. 绑定消息接收后的事件委托
                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body);
                        Console.WriteLine(" [x] Received {0}", message);
                        Thread.Sleep(6000);//模拟耗时
                        Console.WriteLine(" [x] Done");
                    };
                    //7. 启动消费者
                    channel.BasicConsume(queue: "wiky_test_queue", autoAck: true, consumer: consumer);
                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }

我们可以看到客户端代码，相同的地方，都要连接MQ服务，创建channel, 然后申明队列。不同的: 一个是发送消息，一个是接受消息和处理消息，接受消息需要创建消费者实例。

上面的代码可以让我们比较容易地理解生产者，消费者的主要工作流程。

 

当我们了解了上面的基础知识之后，我们在实际开发中，可以对以上的代码进行封装。这个想法，是千千万万开发者的想法，所以有很多的mq代码库。下面我将选用一个最简单的 EasyNetQ。

安装了这个nuget包之后：

先创建一个公开类库，可以被生产者和消费者共享的内容。

定义了一个传输类，一个接口方法用于处理消息

    public class TestEvent
    {
        public int TestId { get; set; }
        public string TestMessage { get; set; }
    }

    public interface ITestSerivce
    {
        void HandleMessage(TestEvent @event);
    }

    

 

生产者代码：

public static IBus Bus { get; set; }
        static void Main(string[] args)
        {
            var rabbitMqConnection = "host=192.168.1.23;virtualHost=/host;username=admin;password=123456";
            Bus = RabbitHutch.CreateBus(rabbitMqConnection);
            Bus.Publish<TestEvent>(new TestEvent()
            {
                TestId = 1,
                TestMessage = "Hello Easy MQ!"
            });

 

消费者代码：

先实现处理消息接口

public class TestService : ITestSerivce
    {
        public void HandleMessage(TestEvent @event)
        {
            Console.WriteLine($"I am handling the event...{@event.TestId}..{@event.TestMessage}...");
            Console.WriteLine("good job!");
        }
    }

下面是消费队列：

 

 public static IBus Bus { get; set; }
        static void Main(string[] args)
        {
            var rabbitMqConnection = "host=192.168.1.23;virtualHost=/host;username=admin;password=123456";
            Bus = RabbitHutch.CreateBus(rabbitMqConnection);
            ITestSerivce serivce = new TestService();
            Bus.Subscribe<TestEvent>(typeof(ITestSerivce).ToString(), serivce.HandleMessage);
            Console.WriteLine("Listening for test message. Enter <exit> to quit.");
            Console.Read();
        }

 

可以看到代码简单很多，变成了一个纯粹的发送消息，处理消息，采用了事件总线的模式。