RabbitMQ消息队列随笔

本文权当各位看官对RabbitMQ的基本概念以及使用场景有了一定的了解，如果你还对它所知甚少或者只是停留在仅仅是听说过，建议你先看看这篇文章，在对RabbitMQ有了基本认识后，我们正式开启我们的RabbitMQ之旅吧，希望本文能够帮助大家在实际用到消息队列时有所帮助，如有表述的不当之处，还望各位看官指正。

一、消息队列的安装

1、 RabbitMQ是用Erlang编程语言进行开发，所以首先得在Erlang官网下载Erlang运行的环境，如图，选择所需对应文件进行下载，并进行安装：

2、 设置环境变量，如图

3、 去RabbitMQ官网下载对应操作系统的安装文件，并进行安装如图：

4、 以管理员方式打开cmd，定位到RabbitMQ安装目录sbin文件夹下，依次执行以下命令

（1）     rabbitmq-service install

（2）     rabbitmq-service enable

（3）     rabbitmq-service start

如图：

到这里我们的RabbitMQ服务已经安装好了，利用Windows + R键，输入services.msc查看，RabbitMQ服务已经处于运行的状态了。

5、 到这里不要以为我们的工作就结束了，接下来我们为RabbitMQ设置用户以及密码

在cmd中执行 rabbitmqctl list_users 查看RabbitMQ已存在的用户，如图

这里发现有两个账号 rabbit 是我之前添加的，guest 是RabbitMQ自带的，

执行以下命令，添加RabbitMQ用户，并设置相应权限:

rabbitmqctl add_user bestadmin 123456

rabbitmqctl set_permissions  bestadmin ".*"  ".*"  ".*"

rabbitmqctl set_user_tags bestadmin administrator

如图：

6、 RabbitMQ有一个可视化界面，进行消息的管理，不过需要用命名rabbitmq-plugins enable rabbitmq_management 命令进行启动，接下来我们便可以在浏览器输入127.0.0.1:15672中进行查看，如图：

输入我们刚设置的用户名bestuser 密码123，如图：

一、消息队列的使用

我们知道RabbitMQ的Exchange常用交换器类型分为fanout、direct、topic、headers 4种类型，这里我们将对fanout、direct、topic 3种类型以实际代码的形式进行讲解，至于关于交换器对各类型的具体讲解，请参照文章开始给出的链接进行了解，这里就不再赘述，我们新建了如下图的解决方案：

1、RabbitMQHelper 帮助类，对常用的消息入队以及消费消息进行了简单的封装：

 /// <summary>
    /// RabbitMQHelper
    /// </summary>
    public class RabbitMQHelper
    {
        private static ConnectionFactory _connectionFactory = null;
        private static readonly JsonSerializerSettings _jsonSettings = new JsonSerializerSettings { Formatting = Formatting.None, NullValueHandling = NullValueHandling.Ignore };

        /// <summary>
        /// 构造函数
        /// </summary>
        static RabbitMQHelper()
        {
            _connectionFactory = new ConnectionFactory();
            _connectionFactory.HostName = ConfigurationManager.AppSettings["HostName"].ToString();
            _connectionFactory.UserName = ConfigurationManager.AppSettings["UserName"].ToString();
            _connectionFactory.Password = ConfigurationManager.AppSettings["Password"].ToString();
            _connectionFactory.AutomaticRecoveryEnabled = true;
        }

        #region 单消息入队
        /// <summary>
        /// 单消息入队
        /// </summary>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="message">消息实例</param>
        /// <param name="arguments">消息的参数信息(如: 队列过期时间:x-expires；消息过期时间:x-message-ttl；过期消息转向路由:x-dead-letter-exchange；过期消息转向路由:x-dead-letter-routing-key 等。)</param>
        public static void Enqueue<TItem>(string exchangeName, string exchangeType, string routingKey, string queueName, TItem message, IDictionary<string, object> arguments = null)
        {
            if (message != null)
            {
                using (IConnection connection = _connectionFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchangeName, exchangeType, true, false, arguments);
                        channel.QueueDeclare(queueName, true, false, false, arguments);
                        channel.QueueBind(queueName, exchangeName, routingKey);
                        var properties = channel.CreateBasicProperties();
                        properties.Persistent = true; //使消息持久化
                        properties.ContentType = "application/json";
                        string messageString = JsonConvert.SerializeObject(message, _jsonSettings);
                        byte[] body = Encoding.UTF8.GetBytes(messageString);
                        channel.BasicPublish(exchangeName, routingKey, properties, body);
                    }
                }
            }
        }
        #endregion

        #region 单消息入队(字符串入队)
        /// <summary>
        /// 单消息入队(字符串入队)
        /// </summary>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="message">消息实例</param>
        /// <param name="arguments">消息的参数信息(如: 队列过期时间:x-expires；消息过期时间:x-message-ttl；过期消息转向路由:x-dead-letter-exchange；过期消息转向路由:x-dead-letter-routing-key 等。)</param>
        public static void Enqueue(string exchangeName, string exchangeType, string routingKey, string queueName, string message, IDictionary<string, object> arguments = null)
        {
            if (!string.IsNullOrWhiteSpace(message))
            {
                using (IConnection connection = _connectionFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchangeName, exchangeType, true, false, arguments);
                        channel.QueueDeclare(queueName, true, false, false, arguments);
                        channel.QueueBind(queueName, exchangeName, routingKey);
                        var properties = channel.CreateBasicProperties();
                        properties.Persistent = true; //使消息持久化
                        byte[] body = Encoding.UTF8.GetBytes(message);
                        channel.BasicPublish(exchangeName, routingKey, properties, body);
                    }
                }
            }
        }
        #endregion

        #region 消息批量入队
        /// <summary>
        /// 消息批量入队
        /// </summary>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="list">消息集合</param>
        /// <param name="arguments">消息的参数信息(如: 队列过期时间:x-expires；消息过期时间:x-message-ttl；过期消息转向路由:x-dead-letter-exchange；过期消息转向路由:x-dead-letter-routing-key 等。)</param>
        public static void Enqueue<TItem>(string exchangeName, string exchangeType, string routingKey, string queueName, List<TItem> list, IDictionary<string, object> arguments = null)
        {
            if (list != null && list.Count > 0)
            {
                using (IConnection connection = _connectionFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        foreach (TItem item in list)
                        {
                            if (item != null)
                            {
                                channel.ExchangeDeclare(exchangeName, exchangeType, true, false, arguments);
                                channel.QueueDeclare(queueName, true, false, false, arguments);
                                channel.QueueBind(queueName, exchangeName, routingKey);
                                string messageString = JsonConvert.SerializeObject(item, _jsonSettings);
                                byte[] body = Encoding.UTF8.GetBytes(messageString);
                                var properties = channel.CreateBasicProperties();//使消息持久化
                                properties.ContentType = "application/json";
                                properties.Persistent = true;
                                channel.BasicPublish(exchangeName, routingKey, properties, body);
                            }
                        }
                    }
                }
            }
        }
        #endregion

        #region 消费消息队列(旧的方式)
        /// <summary>
        /// 消费消息队列
        /// </summary>
        /// <typeparam name="TItem">消息对象</typeparam>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="func">消费消息的具体操作</param>
        /// <param name="failFunc">消费消息失败的具体操作</param>
        /// <param name="tryTimes">消费失败后，继续尝试消费的次数</param>
        /// <param name="arguments">消息的参数信息(如: 队列过期时间:x-expires；消息过期时间:x-message-ttl；过期消息转向路由:x-dead-letter-exchange；过期消息转向路由:x-dead-letter-routing-key 等。)</param>
        /// <param name="isAgain">是否重新入队</param>
        public static void Consume<TItem>(string exchangeName, string exchangeType, string routingKey, string queueName, Func<TItem, bool> func, Func<TItem, bool> failFunc = null,
            int tryTimes = 5, IDictionary<string, object> arguments = null, bool isAgain = false)
        {
            try
            {
                int consumeCount = 0;//尝试消费次数
                bool isConsumeSuccess;//是否消费成功
                using (IConnection connection = _connectionFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchangeName, exchangeType, true, false, arguments);
                        channel.QueueDeclare(queueName, true, false, false, arguments);
                        channel.QueueBind(queueName, exchangeName, routingKey);
                        QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
                        channel.BasicConsume(queueName, false, consumer);
                        while (true)
                        {
                            var ea = consumer.Queue.Dequeue();
                            var body = ea.Body;
                            var message = Encoding.UTF8.GetString(body);
                            TItem queueMessage = JsonConvert.DeserializeObject<TItem>(message, _jsonSettings);
                            if (queueMessage != null)
                            {
                                consumeCount = 0;
                                while (true)
                                {
                                    consumeCount++;
                                    isConsumeSuccess = func(queueMessage);
                                    if (isConsumeSuccess || consumeCount >= tryTimes)
                                    {
                                        channel.BasicAck(ea.DeliveryTag, false);//将队列里面的消息进行释放
                                        if (!isConsumeSuccess && failFunc != null)
                                        {
                                            failFunc(queueMessage);//消费消息失败的具体操作
                                        }
                                        #region 消息处理失败后重新入队
                                        if (!isConsumeSuccess && isAgain)
                                        {
                                            var properties = channel.CreateBasicProperties();
                                            properties.Persistent = true;
                                            channel.BasicPublish(exchangeName, routingKey, properties, body);
                                        }
                                        #endregion
                                        break;
                                    }
                                }
                            }
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
        #endregion

        #region 消费消息队列(旧的方式)
        /// <summary>
        /// 消费消息队列
        /// </summary>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="func">消费消息的具体操作</param>
        /// <param name="failFunc">消费消息失败的具体操作</param>
        /// <param name="tryTimes">消费失败后，继续尝试消费的次数</param>
        /// <param name="arguments">消息的参数信息(如: 队列过期时间:x-expires；消息过期时间:x-message-ttl；过期消息转向路由:x-dead-letter-exchange；过期消息转向路由:x-dead-letter-routing-key 等。)</param>
        /// <param name="isAgain">是否重新入队</param>
        public static void ConsumeString(string exchangeName, string exchangeType, string routingKey, string queueName, Func<string, bool> func, Func<string, bool> failFunc = null,
            int tryTimes = 5, IDictionary<string, object> arguments = null, bool isAgain = false)
        {
            try
            {
                int consumeCount = 0;//尝试消费次数
                using (IConnection connection = _connectionFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchangeName, exchangeType, true, false, arguments);
                        channel.QueueDeclare(queueName, true, false, false, arguments);
                        channel.QueueBind(queueName, exchangeName, routingKey);
                        QueueingBasicConsumer consumer = new QueueingBasicConsumer(channel);
                        channel.BasicConsume(queueName, false, consumer);
                        while (true)
                        {
                            var ea = consumer.Queue.Dequeue();
                            var body = ea.Body;
                            var message = Encoding.UTF8.GetString(body);
                            if (!string.IsNullOrWhiteSpace(message))
                            {
                                consumeCount = 0;
                                while (true)
                                {
                                    consumeCount++;
                                    bool isConsumeSuccess = func(message);
                                    if (isConsumeSuccess || consumeCount >= tryTimes)
                                    {
                                        channel.BasicAck(ea.DeliveryTag, false);//将队列里面的消息进行释放
                                        if (!isConsumeSuccess && failFunc != null)
                                        {
                                            failFunc(message);//消费消息失败的具体操作
                                        }
                                        #region 消息处理失败后重新入队
                                        if (!isConsumeSuccess && isAgain)
                                        {
                                            var properties = channel.CreateBasicProperties();
                                            properties.Persistent = true;
                                            channel.BasicPublish(exchangeName, routingKey, properties, body);
                                        }
                                        #endregion
                                        break;
                                    }
                                }
                            }
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
        #endregion

        #region 消费消息队列(新的方式)
        /// <summary>
        /// 消费消息队列
        /// </summary>
        /// <typeparam name="TItem">消息对象</typeparam>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="func">消费消息的具体操作</param>
        /// <param name="tryTimes">消费失败后，继续尝试消费的次数</param>
        /// <param name="arguments">消息的参数信息(如: 队列过期时间:x-expires；消息过期时间:x-message-ttl；过期消息转向路由:x-dead-letter-exchange；过期消息转向路由:x-dead-letter-routing-key 等。)</param>
        public static void NewConsume<TItem>(string exchangeName, string exchangeType, string routingKey, string queueName, Func<TItem, bool> func, int tryTimes = 5, IDictionary<string, object> arguments = null)
        {
            try
            {
                int consumeCount = 0;//尝试消费次数
                using (IConnection connection = _connectionFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchangeName, exchangeType, true, false, arguments);
                        channel.QueueDeclare(queueName, true, false, false, arguments);
                        channel.QueueBind(queueName, exchangeName, routingKey);
                        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                        consumer.Received += (sender, eventArgs) =>
                        {
                            byte[] body = eventArgs.Body;
                            if (body != null && body.Length > 0)
                            {
                                string message = Encoding.UTF8.GetString(body);
                                if (!string.IsNullOrWhiteSpace(message))
                                {
                                    TItem queueMessage = JsonConvert.DeserializeObject<TItem>(message, _jsonSettings);
                                    if (queueMessage != null)
                                    {
                                        consumeCount = 0;
                                        while (true)
                                        {
                                            consumeCount++;
                                            bool isConsumeSuccess = func(queueMessage);
                                            if (isConsumeSuccess || consumeCount >= tryTimes)
                                            {
                                                channel.BasicAck(eventArgs.DeliveryTag, false);//将队列里面的消息进行释放
                                                break;
                                            }
                                        }
                                    }
                                }
                            }
                        };
                        channel.BasicConsume(queueName, false, consumer);
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
        #endregion

        #region 消费消息队列(新的方式)
        /// <summary>
        /// 消费消息队列
        /// </summary>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="exchangeType">交换器类型</param>
        /// <param name="routingKey">路由关键字</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="func">消费消息的具体操作</param>
        /// <param name="failFunc">消费消息失败的具体操作</param>
        /// <param name="tryTimes">消费失败后，继续尝试消费的次数</param>
        /// <param name="arguments">消息的参数信息(如: 队列过期时间:x-expires；消息过期时间:x-message-ttl；过期消息转向路由:x-dead-letter-exchange；过期消息转向路由:x-dead-letter-routing-key 等。)</param>
        /// <param name="isAgain">是否重新入队</param>
        public static void NewConsumeString(string exchangeName, string exchangeType, string routingKey, string queueName, Func<string, bool> func, Func<string, bool> failFunc = null,
            int tryTimes = 5, IDictionary<string, object> arguments = null, bool isAgain = false)
        {
            try
            {
                int consumeCount = 0;//尝试消费次数
                using (IConnection connection = _connectionFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchangeName, exchangeType, true, false, arguments);
                        channel.QueueDeclare(queueName, true, false, false, arguments);
                        channel.QueueBind(queueName, exchangeName, routingKey);
                        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                        consumer.Received += (sender, eventArgs) =>
                        {
                            byte[] body = eventArgs.Body;
                            if (body != null && body.Length > 0)
                            {
                                string message = Encoding.UTF8.GetString(body);
                                if (!string.IsNullOrWhiteSpace(message))
                                {
                                    consumeCount = 0;
                                    while (true)
                                    {
                                        consumeCount++;
                                        bool isConsumeSuccess = func(message);
                                        if (isConsumeSuccess || consumeCount >= tryTimes)
                                        {
                                            channel.BasicAck(eventArgs.DeliveryTag, false);//将队列里面的消息进行释放
                                            if (!isConsumeSuccess && failFunc != null)
                                            {
                                                failFunc(message);//消费消息失败的具体操作
                                            }
                                            #region 消息处理失败后重新入队
                                            if (!isConsumeSuccess && isAgain)
                                            {
                                                var properties = channel.CreateBasicProperties();
                                                properties.Persistent = true;
                                                channel.BasicPublish(exchangeName, routingKey, properties, body);
                                            }
                                            #endregion
                                            break;
                                        }
                                    }
                                }
                            }
                        };
                        channel.BasicConsume(queueName, false, consumer);
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
        #endregion
    }

 备注：需要说明的是这里在对消费消息的方法进行封装的过程中使用了泛型委托，这样我们就只需要按照自己的业务需求，对消息进行处理了。

 2、 fanout类型：简单的说适合这样的场景，一个生产者产生的消息，需要将该消息发送到多个消息队列，供多个消费者进行消费。这里，为了对该场景进行还原，所以新建了RabbitMQConsumer，RabbitMQConsumer1两个消费者。

 生产者代码：

       public Form1()
        {
            InitializeComponent();
        }

        private void Producer_Click(object sender, EventArgs e)
        {
            RabbitMQHelper.Enqueue("developExchange", "fanout", "", new Developer() { Id = Guid.NewGuid(), Name = "nickdeng", Position = "开发" });
        }

 消费者1代码：

  class Program
    {
        static void Main(string[] args)
        {
            RabbitMQHelper.Consume<Developer>("developExchange", "fanout", "", "developQueue", ConsumeMessage);
        }

        /// <summary>
        /// 消费消息
        /// </summary>
        /// <param name="developer">处理对象</param>
        /// <returns>消费结果</returns>
        public static bool ConsumeMessage(Developer developer)
        {
            string message = JsonConvert.SerializeObject(developer);
            Console.Write(message);
            return true;
        }
    }

消费者2代码：

  class Program
    {
        static void Main(string[] args)
        {
            RabbitMQHelper.Consume<Developer>("developExchange", "fanout", "", "developQueue1", ConsumeMessage);
        }

        /// <summary>
        /// 消费消息
        /// </summary>
        /// <param name="developer">处理对象</param>
        /// <returns>消费结果</returns>
        public static bool ConsumeMessage(Developer developer)
        {
            string message = JsonConvert.SerializeObject(developer);
            Console.Write(message);
            return true;
        }
    }

消费者1与消费者2的代码，眨眼一看，不是一样的吗？仔细看会发现它们在的消息队列名称不一样，消费者1的队列名称是“developQueue”，消息者2的队列名称是“developQueue1”，因为这两个消息队列都与交换器“developExchange”进行了绑定，所以生产者产生的消息将被推送到这两个消息队列。运行代码，得到如下图结果：

 3、direct类型：直译过来就是直接的意思，该类型适用于点对点的使用场景，生产者将消息发送到指定的消息队列：

  生产者代码：

      public Form1()
        {
            InitializeComponent();
        }

        private void Producer_Click(object sender, EventArgs e)
        {
            RabbitMQHelper.Enqueue("developExchange1", "direct", "directkey", new Developer() { Id = Guid.NewGuid(), Name = "nickdeng", Position = "开发" });
        }

消费者1代码：

        static void Main(string[] args)
        {
            RabbitMQHelper.Consume<Developer>("developExchange1", "direct", "directkey", "developQueue", ConsumeMessage);
        }

        /// <summary>
        /// 消费消息
        /// </summary>
        /// <param name="developer">处理对象</param>
        /// <returns>消费结果</returns>
        public static bool ConsumeMessage(Developer developer)
        {
            string message = JsonConvert.SerializeObject(developer);
            Console.Write(message);
            return true;
        }

消费者2代码：

        static void Main(string[] args)
        {
            RabbitMQHelper.Consume<Developer>("developExchange1", "direct", "directkey1", "developQueue1", ConsumeMessage);
        }

        /// <summary>
        /// 消费消息
        /// </summary>
        /// <param name="developer">处理对象</param>
        /// <returns>消费结果</returns>
        public static bool ConsumeMessage(Developer developer)
        {
            string message = JsonConvert.SerializeObject(developer);
            Console.Write(message);
            return true;
        }

生产者的路由关键字是“directkey”，消费者1的路由关键字为”directkey“，消费者2的路由关键字为”directkey1“，仅仅一字相差，生产者产生的消息就只有消费者1能够收到，运行代码得到如图结果：

 至于topic类型，这里就不再以代码进行讲解了，其实大致使用方法与上面的direct类型相似，不同之处在于topic类型可通过路由关键字进行模糊匹配，将消息路由到相应队列，大家可根据自己的实际使用场景，进行类型的选择。