C# 封裝RabbitMQ消息隊列處理

現在使用.net領域使用RabbitMQ有很多解決方案，我自己使用過的有兩個，一個是EasyNetQ，一個是CAP，都挺好用的，尤其是CAP，懶人推薦使用，怎么使用的文章也很多，小伙伴可以自行搜索。

最近我自己嘗試根據目前手頭項目的需要，自行封裝一下基於RabbitMQ的使用，下面開搞，貼上我自己寫的代碼。

首先定義消息發布者/生產者接口：

using System.Threading.Tasks;

namespace fx.MQ
{
    public interface IPublisher
    {
        /// <summary>
        /// 釋放資源。
        /// </summary>
        void Dispose();
        /// <summary>
        /// 
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="message"></param>
        void Publish<T>(T message) where T : class;
        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        /// <param name="channelName"></param>
        void Publish(string message, string channelName);
        /// <summary>
        /// 
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="message"></param>
        /// <returns></returns>
        Task PublishAsync<T>(T message) where T : class;
    }
}

定義訂閱者/消費者接口：

using System;
using System.Threading.Tasks;

namespace fx.MQ
{
    public interface ISubscriber
    {
        /// <summary>
        /// 
        /// </summary>
        void Dispose();
        /// <summary>
        /// 
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="channelName"></param>
        /// <returns></returns>
        void Subscribe(string channelName, Action<string> callback);
        /// <summary>
        /// 
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="channelName"></param>
        /// <returns></returns>
        Task<T> SubscribeAsync<T>(string channelName) where T : class;
    }
}

定義RabbmitMQProvider

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;

namespace fx.MQ
{
    public class RabbitMQProvider
    {
        private readonly string _ipAddress;
        private readonly int? _port;
        private readonly string _username;
        private readonly string _password;

        public RabbitMQProvider(string ipAddress, int? port, string username, string password)
        {
            _ipAddress = ipAddress ?? throw new ArgumentException("IP地址不能為空！");
            _port = port ?? throw new ArgumentException("端口不能為空");
            _username = username ?? throw new ArgumentException("用戶名不能為空");
            _password = password ?? throw new ArgumentException("密碼不能為空");

            ConnectionFactory = new ConnectionFactory//創建連接工廠對象
            {
                HostName = _ipAddress,//IP地址
                Port = (int)_port,//端口號
                UserName = _username,//用戶賬號
                Password = _password//用戶密碼
            };
        }

        public IConnectionFactory ConnectionFactory { get; }

    }
}

實現生產者：

using Newtonsoft.Json;
using RabbitMQ.Client;
using System;
using System.Text;
using System.Threading.Tasks;

namespace fx.MQ
{
    /// <summary>
    /// 消息發布者。
    /// </summary>
    public class RabbitMQPublisher : IPublisher
    {

        private readonly RabbitMQProvider _provider;
        private IConnection _connection;
        public RabbitMQPublisher(RabbitMQProvider provider)
        {
            _provider = provider;
            _connection = _provider.ConnectionFactory.CreateConnection();
        }

        public IConnection Connection
        {
            get
            {
                if (_connection != null)
                    return _connection;
                return _connection = _provider.ConnectionFactory.CreateConnection();
            }
        }

        private IModel _channel;
        public IModel Channel
        {
            get
            {
                if (_channel != null)
                    return _channel;
                else
                    return _channel = _connection.CreateModel();
            }
        }

        public void Dispose()
        {
            if (Channel != null)
            {
                if (Channel.IsOpen)
                    Channel.Close();
                Channel.Abort();
                Channel.Dispose();
            }

            if (Connection != null)
            {
                if (Connection.IsOpen)
                    Connection.Close();
            }
        }

        public void Publish<T>(T message) where T : class
        {
            var channelName = typeof(T).Name;
            Channel.ExchangeDeclare(exchange: channelName, type: "fanout", durable: false, autoDelete: false, null);

            var msgContent = JsonConvert.SerializeObject(message);
            var msgByte = Encoding.UTF8.GetBytes(msgContent);
            Channel.BasicPublish
            (
                exchange: channelName,
                routingKey: string.Empty,
                mandatory: false,
                basicProperties: null,
                body: msgByte
            );
        }


        public void Publish(string message, string channelName)
        {
            Channel.ExchangeDeclare(exchange: channelName, type: "fanout", durable: false, autoDelete: false, null);

            var msgByte = Encoding.UTF8.GetBytes(message);
            Channel.BasicPublish
            (
                exchange: channelName,
                routingKey: string.Empty,
                mandatory: false,
                basicProperties: null,
                body: msgByte
            );
        }

        public Task PublishAsync<T>(T message) where T : class
        {
            throw new NotImplementedException();
        }
    }
}

實現消費者：

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace fx.MQ
{
    /// <summary>
    /// 消息訂閱者/消費者。
    /// </summary>
    public class RabbitMQSubscriber : ISubscriber
    {
        private readonly RabbitMQProvider _provider;
        private IConnection _connection;
        public RabbitMQSubscriber(RabbitMQProvider provider)
        {
            _provider = provider;
            _connection = _provider.ConnectionFactory.CreateConnection();
        }

        public IConnection Connection
        {
            get
            {
                if (_connection != null)
                    return _connection;
                return _connection = _provider.ConnectionFactory.CreateConnection();
            }
        }

        private IModel _channel;
        public IModel Channel
        {
            get
            {
                if (_channel != null)
                    return _channel;
                else
                    return _channel = _connection.CreateModel();
            }
        }


        public void Dispose()
        {
            if (_channel != null)
            {
                _channel.Abort();
                if (_channel.IsOpen)
                    _channel.Close();
                
                _channel.Dispose();
            }

            if (_connection != null)
            {
                if (_connection.IsOpen)
                    _connection.Close();

                _connection.Dispose();
            }
        }

        /// <summary>
        /// 消費消息，並執行回調。
        /// </summary>
        /// <param name="channelName"></param>
        /// <param name="callback"></param>
        public void Subscribe(string channelName, Action<string> callback)
        {
            //聲明交換機
            Channel.ExchangeDeclare(exchange: channelName, type: "fanout");
            //消息隊列名稱
            var queueName = channelName + "_" + Guid.NewGuid().ToString().Replace("-", "");
            //聲明隊列
            Channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
            //將隊列與交換機進行綁定
            Channel.QueueBind(queue: queueName, exchange: channelName, routingKey: "");
            //聲明為手動確認，每次只消費1條消息。
            Channel.BasicQos(0, 1, false);
            //定義消費者
            var consumer = new EventingBasicConsumer(Channel);
            //接收事件
            consumer.Received += (eventSender, args) =>
            {
                var message = args.Body;//接收到的消息

                callback(Encoding.UTF8.GetString(message));
                //返回消息確認
                Channel.BasicAck(args.DeliveryTag, true);
            };
            //開啟監聽
            Channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);

        }

        public Task<T> SubscribeAsync<T>(string channelName) where T : class
        {
            throw new NotImplementedException();
        }
    }
}

到這里為止，簡單的實現消息隊列的接受，發送，已經滿足我自己當前項目的需要了。這里我用的exchange進行消息隊列的生產消費，並且用fanout模式，就是一個生產者對應多個消費者，有點類似於消息廣播，另外還有兩種模式，可以根據需要修改。

下面是測試代碼：

using System;
using System.Windows.Forms;

namespace fx.MQ.TestForm
{
    public partial class Form1 : Form
    {
        private readonly RabbitMQProvider _provider;
        private readonly RabbitMQPublisher _publisher;
        private readonly RabbitMQSubscriber _subscriber;
        delegate void Callback(string msg);

        public Form1()
        {
            _provider = new RabbitMQProvider("192.168.101.199", 5672, "admin", "admin");
            _publisher = new RabbitMQPublisher(_provider);
            _subscriber = new RabbitMQSubscriber(_provider);
            //callback = new Callback(ShowMessage);
            InitializeComponent();
        }

        private void button1_Click(object sender, EventArgs e)
        {
            _publisher.Publish(textBox1.Text, "public");
        }

        private void Form1_Load(object sender, EventArgs e)
        {
            
            _subscriber.Subscribe("public", c=> {
                ShowMessage(c);
            });
        }


        private void ShowMessage(string msg)
        {
            if (this.richTextBox1.InvokeRequired)
            {
                var cb = new Callback(ShowMessage);
                this.Invoke(cb, new object[] { msg });
            }
            else
            {
                this.richTextBox1.Text = msg;
            }
        }
    }
}

運行效果如圖所示：

 

 

OK，沒有問題。

另外注意，退出程序時消息發布者和訂閱者都需要Dispose()來釋放連接。