EasyNetQ自定义异常消息处理

浏览数：220 / 时间：2015年06月09日

　　这几天在折腾消息队列，在.Net环境下有基于RabbitMQ有很多有API的选择，最后选择了比较简单的EasyNetQ(http://easynetq.com/)

　　在测试使用的时候发现一个问题，对于处理错误的消息，EasyNetQ默认会放到一个错误队列中并提供了一个工具可以对错误队列的消息进行重新分发。RabiitMQ是有一个事务功能的，但是貌似在EasyNetQ的实现中，没有发现相关的操作方式

　　现在的需求是，在某种特定的情况下，需要将处理不成功的消息，保留在消息原队列

　　做了一个变通处理，出现错误消息的时候，将其再送回原队列。从客户端上可以直接send回队列，但这样不是一个很好的方式

　　研究了一下EasyNetQ的相关代码，发现其定义了一个IConsumerErrorStrategy接口，我们只要自己自行实现，并注册一个自定义方法就可以

　　ComponentRegistration.cs原注册方法相关，默认错误消息处理方式在DefaultConsumerErrorStrategy.cs中

 1     public class ComponentRegistration
 2     {
 3         public static void RegisterServices(IContainer container)
 4         {
 5             Preconditions.CheckNotNull(container, "container");
 6 
 7             // Note: IConnectionConfiguration gets registered when MQContext.CreateBus(..) is run.
 8             #region DefaultConsumerErrorStrategy
 9             container
10             .Register(_ => container)
11             .Register<IMessageQueueQLogger, ConsoleLogger>()
12             .Register<ISerializer, JsonSerializerN>()
13             .Register<IConventions, Conventions>()
14             .Register<IEventBus, EventBus>()
15             .Register<ITypeNameSerializer, TypeNameSerializer>()
16             .Register<Func<string>>(x => CorrelationIdGenerator.GetCorrelationId)
17             .Register<IClusterHostSelectionStrategy<ConnectionFactoryInfo>, DefaultClusterHostSelectionStrategy<ConnectionFactoryInfo>>()
18             .Register<IConsumerDispatcherFactory, ConsumerDispatcherFactory>()
19             .Register<IPublishExchangeDeclareStrategy, PublishExchangeDeclareStrategy>()
20             .Register<IPublisherConfirms, PublisherConfirms>()
21             .Register<IConsumerErrorStrategy, DefaultConsumerErrorStrategy>()
22             .Register<IHandlerRunner, HandlerRunner>()
23             .Register<IInternalConsumerFactory, InternalConsumerFactory>()
24             .Register<IConsumerFactory, ConsumerFactory>()
25             .Register<IConnectionFactory, ConnectionFactoryWrapper>()
26             .Register<IPersistentChannelFactory, PersistentChannelFactory>()
27             .Register<IClientCommandDispatcherFactory, ClientCommandDispatcherFactory>()
28             .Register<IHandlerCollectionFactory, HandlerCollectionFactory>()
29             .Register<IAdvancedBus, RabbitAdvancedBus>()
30             .Register<IRpc, Rpc>()
31             .Register<ISendReceive, SendReceive>()
32             .Register<IBus, RabbitBus>();
33             #endregion
34         }
35     }

View Code

　　由于最终消息队列中保存的消息都是序列化后的结果，借鉴EasyNetQ自带工具的错误消息处理方式，只需要将消息直接调用BasicPublish方法便可

　　自定义错误消息处理HandleConsumerError方法

 1         public virtual PostExceptionAckStrategy HandleConsumerError(ConsumerExecutionContext context, Exception exception)
 2         {
 3             Preconditions.CheckNotNull(context, "context");
 4             Preconditions.CheckNotNull(exception, "exception");
 5 
 6             try
 7             {
 8                 Connect();
 9 
10                 using (var model = connection.CreateModel())
11                 {
12                     var messageBody = context.Body;
13                     var properties = model.CreateBasicProperties();
14                     context.Properties.CopyTo(properties);
15 
16                     model.BasicPublish(String.Empty, context.Info.RoutingKey, properties, messageBody);
17                 }
18             }          
19             catch (Exception unexpectedException)
20             {
21                 // Something else unexpected has gone wrong :(
22                 logger.ErrorWrite("EasyNetQ Consumer Error Handler: Failed to publish error message\nException is:\n"
23                     + unexpectedException);
24             }
25             return Consumer.PostExceptionAckStrategy.ShouldAck;
26         }

View Code

　　关于原队列的名字保存在RoutingKey属性中

　　IConsumerErrorStrategy实现的完整代码

 1 using System;
 2 using System.Collections.Concurrent;
 3 using System.Text;
 4 using EasyNetQ.SystemMessages;
 5 using RabbitMQ.Client;
 6 using RabbitMQ.Client.Exceptions;
 7 
 8 namespace EasyNetQ.Consumer
 9 {
10     public class DefaultConsumerErrorStrategy : IConsumerErrorStrategy
11     {
12         private readonly IConnectionFactory connectionFactory;
13         private readonly ISerializer serializer;
14         private readonly IMessageQueueQLogger logger;
15         private readonly IConventions conventions;
16         private readonly ITypeNameSerializer typeNameSerializer;
17 
18         private IConnection connection;
19         private readonly ConcurrentDictionary<string, string> errorExchanges = new ConcurrentDictionary<string, string>();
20 
21         public DefaultConsumerErrorStrategy(
22             IConnectionFactory connectionFactory, 
23             ISerializer serializer,
24             IMessageQueueQLogger logger,
25             IConventions conventions, 
26             ITypeNameSerializer typeNameSerializer)
27         {
28             Preconditions.CheckNotNull(connectionFactory, "connectionFactory");
29             Preconditions.CheckNotNull(serializer, "serializer");
30             Preconditions.CheckNotNull(logger, "logger");
31             Preconditions.CheckNotNull(conventions, "conventions");
32             Preconditions.CheckNotNull(typeNameSerializer, "typeNameSerializer");
33 
34             this.connectionFactory = connectionFactory;
35             this.serializer = serializer;
36             this.logger = logger;
37             this.conventions = conventions;
38             this.typeNameSerializer = typeNameSerializer;
39         }
40 
41         private void Connect()
42         {
43             if(connection == null || !connection.IsOpen)
44             {
45                 connection = connectionFactory.CreateConnection();
46             }
47         }
48 
49         public virtual PostExceptionAckStrategy HandleConsumerError(ConsumerExecutionContext context, Exception exception)
50         {
51             Preconditions.CheckNotNull(context, "context");
52             Preconditions.CheckNotNull(exception, "exception");
53 
54             try
55             {
56                 Connect();
57 
58                 using (var model = connection.CreateModel())
59                 {
60                     var messageBody = context.Body;
61                     var properties = model.CreateBasicProperties();
62                     context.Properties.CopyTo(properties);
63 
64                     model.BasicPublish(String.Empty, context.Info.RoutingKey, properties, messageBody);
65                 }
66             }          
67             catch (Exception unexpectedException)
68             {
69                 // Something else unexpected has gone wrong :(
70                 logger.ErrorWrite("EasyNetQ Consumer Error Handler: Failed to publish error message\nException is:\n"
71                     + unexpectedException);
72             }
73             return Consumer.PostExceptionAckStrategy.ShouldAck;
74         }
75 
76         private bool disposed = false;
77 
78         public virtual void Dispose()
79         {
80             if (disposed) return;
81 
82             if(connection != null) connection.Dispose();
83 
84             disposed = true;
85         }
86     }
87 }