RabbitMQ 消息队列 DEMO

Posted 2021-02-03 jasonlai2016

1. 引用 RabbitMQ.Client.5.1.0

2. http://localhost:15672/

 public class TestController : ApiController
    {
        private static string EXCHANGE_NAME = "exchangeTest"; //交换器（路由器），负责消息的路由到相应队列
        private static string QUEUE_NAME = "queueTest";//队列，消息的缓冲存储区
        private static string ROUTINGKEY_Name = "routingKeyTest";//路由键
        private static string MQServer = "localhost";//服务器地址
        private static string MESSAGE = "消息测试";
        /// <summary>
        /// 生产者发送消息
        /// </summary>
        public void MQPublisher()
        {
            //1. 实例化连接工厂
            var factory = new ConnectionFactory() { HostName = MQServer };
            //2. 建立连接
            using (var connection = factory.CreateConnection())
            {
                //3. 创建信道
                using (var channel = connection.CreateModel())
                {
                    //4. 申明队列
                    channel.QueueDeclare(queue: QUEUE_NAME, durable: true, exclusive: false, autoDelete: false, arguments: null);
                    //5. 申明交换器
                    channel.ExchangeDeclare(exchange: EXCHANGE_NAME, type: "direct", durable: true);
                    for (int i = 0; i < 1000; i++)
                    {
                        //6. 构建byte消息数据包
                        var message = JsonConvert.SerializeObject(i.ToString() + MESSAGE, new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All });
                        var body = Encoding.UTF8.GetBytes(message);
                        //7. 发送数据包
                        channel.BasicPublish(exchange: EXCHANGE_NAME, routingKey: ROUTINGKEY_Name, basicProperties: null, body: body);
                    }
                }
            }
        }
        /// <summary>
        /// 消费者接收消息
        /// </summary>
        public void MQConsumer()
        {
            //1.实例化连接工厂
            var factory = new ConnectionFactory() { HostName = MQServer };
            //2. 建立连接
            using (var connection = factory.CreateConnection())
            {
                //3. 创建信道
                using (var channel = connection.CreateModel())
                {
                    //4. 申明队列
                    channel.QueueDeclare(queue: QUEUE_NAME, durable: true, exclusive: false, autoDelete: false, arguments: null);
                    //5. 申明交换器
                    channel.ExchangeDeclare(exchange: EXCHANGE_NAME, type: "direct", durable: true);
                    //6. 构造消费者实例
                    var consumer = new EventingBasicConsumer(channel);
                    //7. 绑定消息接收后的事件委托
                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body);
                    };
                    //8. 启动消费者
                    channel.BasicConsume(queue: QUEUE_NAME, autoAck: true, consumer: consumer);
                }
            }
        }
    }