.net平台RabbitMQ封装及使用

public abstract class MqHelper
    {
        protected static IConnection defaultConnection = null;
        private static ConnectionFactory ConnectionFactory;
        public static string MqConnection = ConfigurationManager.AppSettings["MqConnection"];

        static MqHelper()
        {
            ConnectionFactory = new ConnectionFactory();
            ConnectionFactory.Uri = new Uri(MqConnection);
            ConnectionFactory.AutomaticRecoveryEnabled = true;
            ConnectionFactory.NetworkRecoveryInterval = TimeSpan.FromSeconds(10);
            defaultConnection = ConnectionFactory.CreateConnection();
        }

        /// <summary>
        /// 获取默认的消息队列连接
        /// 今天的参考资料说：“线程间不能共享通道实例”。
        /// </summary>
        public static IConnection DefaultConnection
        {
            get
            {
                if (defaultConnection == null)
                {
                    //读取配置文件
                    string uri = System.Configuration.ConfigurationManager.AppSettings["MqConnection"];
                    defaultConnection = CreateConnectionByUri(uri);

                    #region 例子
                    // factory.Uri ="amqp://guest:guest@localhost:5672/%2f"
                    //factory.HostName = "localhost";   //主机地址
                    //factory.VirtualHost = "/";        //多租户时 虚拟主机地址
                    //factory.Port = 5672;              //服务端口号
                    //factory.UserName = "guest";       //用户名
                    //factory.Password = "guest";       //密码
                    //factory.CreateConnection();
                    #endregion
                }
                return defaultConnection;
            }
        }

        /// <summary>
        /// 通过Uri 创建消息队里连接
        /// </summary>
        /// <param name="uri">amqp://guest:guest@localhost:5672/vhost</param>
        /// <returns></returns>
        public static IConnection CreateConnectionByUri(string uri)
        {
            var factory = new ConnectionFactory();
            factory.Uri = new Uri(uri);
            return factory.CreateConnection();
        }


        /// <summary>
        /// 创建连接
        /// </summary>
        /// <param name="hostName">消息队列服务器地址</param>
        /// <param name="vHost">虚拟主机</param>
        /// <param name="port">端口号</param>
        /// <param name="userName">用户名,  默认值:guest</param>
        /// <param name="password">用户密码,  默认值:guest</param>
        /// <returns></returns>
        /// <example>
        /// using (var connection = MqHelper.CreateConnection()) 
        /// using (var channel = connection.CreateModel())
        /// { 
        ///     ///ToDo:处理
        /// }
        /// </example>
        public static IConnection CreateConnection(string hostName = "", string virtualHost = "/", int port = 5672, string userName = "", string password = "")
        {
            var factory = new ConnectionFactory();
            //主机地址
            factory.HostName = hostName;
            //多租户时 虚拟主机地址
            factory.VirtualHost = virtualHost;
            //端口号
            factory.Port = port;
            //用户名密码为空时从配置文件获取               
            factory.UserName = string.IsNullOrEmpty(userName) ? "guest" : userName;
            factory.Password = string.IsNullOrEmpty(password) ? "guest" : password;

            IConnection connection = factory.CreateConnection();
            return connection;
        }

        /// <summary>
        /// 创建通道
        /// </summary>
        /// <param name="conn"></param>
        /// <returns></returns>
        public static IModel CetateChannel(IConnection conn = null)
        {
            conn = conn ?? DefaultConnection;
            return conn.CreateModel();
        }

        /// <summary>
        /// 声明交换器
        /// </summary>
        /// <param name="exchangeName">交换器名称</param>
        /// <param name="type">类型</param>
        /// <param name="chanel">通道</param>
        /// <param name="durable">是否持久化，默认: true</param>
        /// <param name="autoDelete">自动删除:一旦客户端连接断开则自动删除，默认: false</param>
        /// <param name="arguments">其他参数：如果安装了队列优先级插件则可以设置优先级</param>
        public static void ExchangeDeclare(string exchangeName, string type, IModel chanel = null, bool durable = true, bool autoDelete = false, IDictionary<string, object> arguments = null)
        {
            IModel ch = chanel ?? CetateChannel();
            ch.ExchangeDeclare(exchange: exchangeName,   //队列名
                            type: type,                 // 类型
                            durable: true,              //是否持久化
                            autoDelete: false,          //一旦客户端连接断开则自动删除queue
                            arguments: null);           //如果安装了队列优先级插件则可以设置优先级
            if (chanel == null)
            {
                //销毁自动创建的 通道
                ch.Close();
                ch.Dispose();
            }
        }

        /// <summary>
        /// 声明队列或者查询队列信息
        /// 通过配置加载程序中无需手动声明，队列多次声明不会重复创建
        /// 但是：队列多次被声明,并且配置与已有的不一致时，会抛出 OperationInterruptedException 异常
        /// </summary>
        /// <param name="queueName">队列名称</param>
        /// <param name="channel">与消息队列连接的通道</param>
        /// <param name="durable">是否持久化，默认: true</param>
        /// <param name="exclusive">排它，默认:false </param>
        /// <param name="autoDelete">自动删除:一旦客户端连接断开则自动删除，默认: false</param>
        /// <param name="arguments">其他参数：如果安装了队列优先级插件则可以设置优先级</param>
        public static QueueDeclareOk QueueDeclare(string queueName, IModel channel, bool durable = true, bool exclusive = false, bool autoDelete = false, IDictionary<string, object> arguments = null)
        {
            return channel.QueueDeclare(queue: queueName,   //队列名
                            durable: true,      //是否持久化
                            exclusive: false,   //排它
                            autoDelete: false,  //一旦客户端连接断开则自动删除queue
                            arguments: null);   //如果安装了队列优先级插件则可以设置优先级
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="exchange">交换器</param>
        /// <param name="routingKey">routingKey</param>
        /// <param name="message">消息</param>
        /// <param name="chanel"></param>
        /// <param name="mandatory"></param>
        /// <param name="basicProperties"></param>
        public static void Publish(string exchange, string routingKey, dynamic message, IModel chanel = null, bool mandatory = false, IBasicProperties basicProperties = null)
        {
            IModel ch = chanel ?? CetateChannel();
            var body = MessageUtility.GetBytes(message);
            //ch.BasicPublish(exchange: exchange, routingKey: routingKey, mandatory: mandatory, basicProperties: basicProperties, body:body);
            ch.BasicPublish(exchange, routingKey, mandatory, basicProperties, body);
            //销毁自动创建的通道
            if (chanel == null)
            {
                ch.Close();
                ch.Dispose();
            }
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="exchange">交换器</param>
        /// <param name="routingKey">routingKey</param>
        /// <param name="body">消息内容</param>
        /// <param name="chanel"></param>
        /// <param name="mandatory"></param>
        /// <param name="basicProperties"></param>
        public static void Publish(string exchange, string routingKey, byte[] body, IModel chanel = null, bool mandatory = false, IBasicProperties basicProperties = null)
        {
            IModel ch = (chanel != null) ? chanel : CetateChannel();
            ch.BasicPublish(exchange: exchange, routingKey: routingKey, mandatory: mandatory, basicProperties: basicProperties, body: body);

            //销毁自动创建的通道
            if (chanel == null)
            {
                ch.Close();
                ch.Dispose();
            }
        }
    }

public static class MessageUtility
    {
        /// <summary>
        /// 序列化: 将对象转二进制数组
        /// </summary>
        /// <param name="obj"></param>
        /// <returns></returns>
        public static byte[] GetBytes(dynamic obj)
        {
            if (obj == null)
            {
                return null;
            }
            string json = JsonConvert.SerializeObject(obj);
            return Encoding.UTF8.GetBytes(json);
        }
        /// <summary>
        /// 反序列化:二进制数组转类 
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="messageBody"></param>
        /// <returns></returns>
        public static T ReadBytes<T>(byte[] messageBody)
        {
            if (messageBody == null || messageBody.Length < 1)
            {
                return default(T);
            }
            var json = Encoding.UTF8.GetString(messageBody);
            if (typeof(T) == typeof(string))
            {
                return (T)(object)(json);
            }
            T obj = JsonConvert.DeserializeObject<T>(json);
            return obj;
        }
    }

public static class MqService
    {
        public static Dictionary<string, IModel> Channels = new Dictionary<string, IModel>();
        static Dictionary<string, string> Tags = new Dictionary<string, string>();

        /// <summary>
        /// 注册消息队列接收
        /// </summary>
        /// <param name="routKey">路由</param>
        /// <param name="OnReceived">收到后的消息处理</param>
        public static IModel RegisterConsume(string routKey, EventHandler<BasicDeliverEventArgs> OnReceived, bool autoAck = true)
        {
            IModel sysChannel = Channels.ContainsKey(routKey) ? Channels[routKey] : null;
            if (sysChannel == null || sysChannel.IsClosed)
            {
                sysChannel = MqHelper.CetateChannel();
                EventingBasicConsumer syslConsumer = new EventingBasicConsumer(sysChannel);
                syslConsumer.Received += OnReceived;
                var tag = sysChannel.BasicConsume(routKey, autoAck, syslConsumer);
                if (!Channels.ContainsKey(routKey))
                {
                    Channels.Add(routKey, sysChannel);
                    Tags.Add(routKey, tag);
                }
                else
                {
                    Channels[routKey] = sysChannel;
                    Tags[routKey] = tag;
                }
            }
            return sysChannel;
        }

        /// <summary>
        /// 消息接收以后ACK
        /// </summary>
        /// <param name="e"></param>
        /// <param name="routKey"></param>
        public static void Ack(this BasicDeliverEventArgs e, string routKey)
        {
            if (!Channels.ContainsKey(routKey))
            {
                return;
            }
            IModel sysChannel = Channels[routKey];
            sysChannel.BasicAck(e.DeliveryTag, false);
        }

        ////事件处理的例子
        //private static void SystemLog_OnReceived(object sender, BasicDeliverEventArgs e)
        //{
        //    try
        //    {
        //        IModel sysChannel = Channels[routKey];
        //        channel.basicAck(e.DeliveryTag, false);
        //        SystemLog log = MessageUtility.ReadBytes<SystemLog>(e.Body);
        //        int val = dataService.AddLog(log);
        //        if (val < 1)
        //        {
        //            LogManager.GetLogger(Assembly.GetExecutingAssembly().GetName().Name).Info("忽略系统日志:" + JsonConvert.SerializeObject(log));
        //        }
        //    }
        //    catch (Exception ex)
        //    {
        //        LogManager.GetLogger(Assembly.GetExecutingAssembly().GetName().Name).Error(ex);
        //    }
        //}

        /// <summary>
        /// 关闭 消息队列监听
        /// </summary>
        /// <param name="routKey"></param>
        public static void UnRegister(string routKey)
        {
            IModel chanel = null;
            try
            {
                if (!Channels.ContainsKey(routKey) && !Tags.ContainsKey(routKey))
                {
                    return;
                }
                chanel = Channels[routKey];
                var tag = Tags[routKey];
                chanel.BasicCancel(tag);
                Channels.Remove(routKey);
                Tags.Remove(routKey);
            }
            finally
            {
                chanel?.Close();
            }
        }
    }

调用：

//读取配置文件

public string RoutKey = ConfigurationManager.AppSettings["MQ_Info"];

//接收MQ消息

 MqService.RegisterConsume(RoutKey, OnReceived, false);

private void OnReceived(object sender, BasicDeliverEventArgs e)
{
try
{

//把消息进行格式转换
var model = MessageUtility.ReadBytes<Model>(e.Body);

//确认
MqService.Ack(e, RoutKey);
}
catch (Exception ex)
{
}
}