XYY.Core/1.Command/XYY.Task.Standard/BaseConsumer.cs

using System;
using System.Threading.Tasks;

namespace XYY.TaskTrack.Standard
{
    /// <summary>
    /// 消费者基类，主要是用于统一消费者出现异常时，能通过发送死信队列的方式来捕获及善后处理，而不是因
    /// 消费者出现异常时没有ACK导致消息被持久化并重复消费。
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public abstract class BaseConsumer<T> : IBaseConsumer<T> where T : class
    {
        public IMQManager _mQManager;
        public BaseConsumer(IMQManager mQManager)
        {
            mQManager = _mQManager;
        }

        /// <summary>
        /// 处理消费者出现的异常。当出现异常时，通过该方法发送一条死信消息供后续处理。
        /// </summary>
        /// <param name="ex"></param>
        /// <param name="message"></param>
        protected void PublishDeadMessage(Exception ex, T message)
        {
            DeadMessage<T> dealMessage = new DeadMessage<T>
            {
                Exception = ex,
                Message = message
            };

            _mQManager.PublishDeadMessage(dealMessage);
        }

        /// <summary>
        /// 消费消息
        /// </summary>
        /// <param name="message"></param>
        public abstract Task Consume(T message);
    }
}