SMP.Service/SMP.Message.Consumer/MQConsumers/MQManager.cs

using EasyNetQ;
using EasyNetQ.FluentConfiguration;
using SMP.Message.Consumer.Config;

namespace SMP.Message.Consumer.MQConsumers
{
    public class MQManager
    {
        private static IBus _bus;

        private readonly static object syncRoot = new object();

        private MQManager()
        {

        }

        private static IBus bus
        {
            get
            {
                if (_bus == null)
                {
                    lock (syncRoot)
                    {
                        if (_bus == null)
                        {
                            string connStr = RabbitMqConfig.MqConnection;
                            _bus = RabbitHutch.CreateBus(connStr);
                        }

                    }
                }
                return _bus;
            }
        }

        /// <summary>
        /// 发送一条消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="message"></param>
        public static void Publish<T>(T message) where T : class
        {
            try
            {
                bus.Publish(message);
            }
            catch (Exception ex)
            {
                Log.Log.Error(ex, ex.Message);
            }
        }

        /// <summary>
        /// 异步发送消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="message"></param>
        /// <returns></returns>
        public static Task PublishAsync<T>(T message) where T : class
        {
            return bus.PublishAsync(message);
        }

        /// <summary>
        /// 指定配置项发送消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="message"></param>
        /// <param name="configure"></param>
        public static void Publish<T>(T message, Action<IPublishConfiguration> configure) where T : class
        {
            bus.Publish(message, configure);
        }

        /// <summary>
        /// 订阅消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="consumer"></param>
        public static void Subscribe<T>(BaseConsumer<T> consumer) where T : class
        {
            try
            {
                bus.Subscribe<T>(string.Empty, consumer.Consume);
            }
            catch (Exception ex)
            {
                Log.Log.Error(ex, ex.Message);
            }
        }

        /// <summary>
        /// 指定配置订阅消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="action"></param>
        /// <param name="configure"></param>
        public static void Subscribe<T>(Action<T> action, Action<ISubscriptionConfiguration> configure) where T : class
        {
            try
            {
                bus.Subscribe<T>(string.Empty, action, configure);
            }
            catch (Exception ex)
            {
                Log.Log.Error(ex, ex.Message);
            }
        }

        /// <summary>
        /// 发送一条死信
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="deadMessage"></param>
        public static void PublishDeadMessage<T>(DeadMessage<T> deadMessage) where T : class
        {
            try
            {
                string typeName = typeof(T).FullName;
                string queueName = $"dead-{typeName}";

                var x = bus.Advanced.ExchangeDeclare("Dead_Message_Exchange", "direct");
                var q = bus.Advanced.QueueDeclare(queueName);
                bus.Advanced.Bind(x, q, queueName);

                string routingKey = queueName;
                var msg = new Message<DeadMessage<T>>(deadMessage);
                bus.Advanced.PublishAsync<DeadMessage<T>>(x, routingKey, true, msg);
            }
            catch (Exception ex)
            {
                Log.Log.Error(ex, ex.Message);
            }
        }
    }
}