EasyNetQ自定義異常消息處理

　　20140310補充：

　　rabbitmq有requeue屬性，可以選擇消息是否返回隊列，另，本文的解決方式非常之山寨，只能應用於發送和接收方式。　　

 

　　這幾天在折騰消息隊列，在.Net環境下有基於RabbitMQ有很多有API的選擇，最后選擇了比較簡單的EasyNetQ(http://easynetq.com/)

　　在測試使用的時候發現一個問題，對於處理錯誤的消息，EasyNetQ默認會放到一個錯誤隊列中並提供了一個工具可以對錯誤隊列的消息進行重新分發。RabiitMQ是有一個事務功能的，但是貌似在EasyNetQ的實現中，沒有發現相關的操作方式

　　現在的需求是，在某種特定的情況下，需要將處理不成功的消息，保留在消息原隊列

　　做了一個變通處理，出現錯誤消息的時候，將其再送回原隊列。從客戶端上可以直接send回隊列，但這樣不是一個很好的方式

　　研究了一下EasyNetQ的相關代碼，發現其定義了一個IConsumerErrorStrategy接口，我們只要自己自行實現，並注冊一個自定義方法就可以

　　ComponentRegistration.cs原注冊方法相關，默認錯誤消息處理方式在DefaultConsumerErrorStrategy.cs中

    public class ComponentRegistration
    {
        public static void RegisterServices(IContainer container)
        {
            Preconditions.CheckNotNull(container, "container");

            // Note: IConnectionConfiguration gets registered when MQContext.CreateBus(..) is run.
            #region DefaultConsumerErrorStrategy
            container
            .Register(_ => container)
            .Register<IMessageQueueQLogger, ConsoleLogger>()
            .Register<ISerializer, JsonSerializerN>()
            .Register<IConventions, Conventions>()
            .Register<IEventBus, EventBus>()
            .Register<ITypeNameSerializer, TypeNameSerializer>()
            .Register<Func<string>>(x => CorrelationIdGenerator.GetCorrelationId)
            .Register<IClusterHostSelectionStrategy<ConnectionFactoryInfo>, DefaultClusterHostSelectionStrategy<ConnectionFactoryInfo>>()
            .Register<IConsumerDispatcherFactory, ConsumerDispatcherFactory>()
            .Register<IPublishExchangeDeclareStrategy, PublishExchangeDeclareStrategy>()
            .Register<IPublisherConfirms, PublisherConfirms>()
            .Register<IConsumerErrorStrategy, DefaultConsumerErrorStrategy>()
            .Register<IHandlerRunner, HandlerRunner>()
            .Register<IInternalConsumerFactory, InternalConsumerFactory>()
            .Register<IConsumerFactory, ConsumerFactory>()
            .Register<IConnectionFactory, ConnectionFactoryWrapper>()
            .Register<IPersistentChannelFactory, PersistentChannelFactory>()
            .Register<IClientCommandDispatcherFactory, ClientCommandDispatcherFactory>()
            .Register<IHandlerCollectionFactory, HandlerCollectionFactory>()
            .Register<IAdvancedBus, RabbitAdvancedBus>()
            .Register<IRpc, Rpc>()
            .Register<ISendReceive, SendReceive>()
            .Register<IBus, RabbitBus>();
            #endregion
        }
    }

　　對DefaultConsumerErrorStrategy的代碼進行研究后發現，EasyNetQ是產生一個處理錯誤消息隊列的Exchanges，將消息二次封裝后推送到默認的錯誤隊列

　　只要將消息隊列改為源消息隊列，取消消息二次封裝，直接推送到隊列

　　隊列綁定

        private string DeclareErrorExchangeAndBindToDefaultErrorQueue(IModel model, ConsumerExecutionContext context)
        {
            var originalRoutingKey = context.Info.RoutingKey;

            return errorExchanges.GetOrAdd(originalRoutingKey, _ =>
            {
                var exchangeName = conventions.ErrorExchangeNamingConvention(context.Info);
                model.ExchangeDeclare(exchangeName, ExchangeType.Direct, durable: true);
                //更改第一個參數
                model.QueueBind(originalRoutingKey, exchangeName, originalRoutingKey);
                return exchangeName;
            });
        }

　　消息處理

        public virtual PostExceptionAckStrategy HandleConsumerError(ConsumerExecutionContext context, Exception exception)
        {
            Preconditions.CheckNotNull(context, "context");
            Preconditions.CheckNotNull(exception, "exception");

            try
            {
                Connect();

                using (var model = connection.CreateModel())
                {
                    var errorExchange = DeclareErrorExchangeQueueStructure(model, context);
                    var messageBody = context.Body;
                    var properties = model.CreateBasicProperties();
                    context.Properties.CopyTo(properties);
                    properties.Type = context.Properties.Type;
                    //消息持久化
                    properties.SetPersistent(true);

                    model.BasicPublish(errorExchange, context.Info.RoutingKey, properties, messageBody);

                }
            }          
            catch (Exception unexpectedException)
            {
                // Something else unexpected has gone wrong :(
                logger.ErrorWrite("EasyNetQMessageQueue Consumer Error Handler: Failed to publish error message\nException is:\n"
                    + unexpectedException);
            }
            return Consumer.PostExceptionAckStrategy.ShouldAck;
        }

　　因為EasyNetQ在CreateBus的時候就會將相關事件注冊，所以我們只需最后自行在config中加入有關配置，在注冊事件的位置加入判斷即可讓異常拋回

　　由於剛接觸RabbitMQ，對於一些概念的理解可能還不是非常到位，這是我目前所能找到的解決方案

　　當然可能以后的版本中，EasyNetQ會加入對事務的操作，現在EasyNetQ支持了一種叫Publisher Confirms的方法，貌似還是不是我想要的