EasyNetQ异常处理

代码下载

https://download.csdn.net/download/u010312811/11252093

官方Demo

https://github.com/EasyNetQ/EasyNetQ/issues/793

 

1.新建项目

创建一个控制台程序，并添加对 EasyNetQ的引用

 

2.创建消息模型

2.1创建Answer

创建Answer数据模型

    public class Answer 2  { 3         public string Text { get; } 4 
        public Answer(string text) 6  { 7             Text = text; 8  } 9     }

2.2创建Question

    public class Question 2  { 3         public string Text { get; } 4 
        public Question(string text) 6  { 7             Text = text; 8  } 9     }

 

3.测试程序

3.1初始化代码

   private static IBus bus;  2    private const string ErrorQueue = "EasyNetQ_Default_Error_Queue";  3 
   static void Main(string[] args)  5  {  6         bus = RabbitHutch.CreateBus("host=localhost");  7         /*订阅消息*/
 Subscribe();  9 
        /*处理错误队列中的错误数据*/
 HandleErrors(); 12 
        /*发布消息*/
        Console.WriteLine("输入文字，按回车发送消息！"); 15         while (true) 16  { 17             var msg = Console.ReadLine(); 18             bus.Publish(new Question(msg)); 19  } 20     }

创建一个总线，用于消息的收发；

依次注册消息的订阅方法，错误处理方法，消息发布方法。

 

3.2消息订阅

    private static void Subscribe()  2  {  3         /*声明两个消费者*/
        bus.SubscribeAsync<Question>("subscriptionId", x => HandleMessageAsync(x).Invoke(1));  5         bus.SubscribeAsync<Question>("subscriptionId", x => HandleMessageAsync(x).Invoke(2));  6  }  7 
    private static Func<int,Task> HandleMessageAsync(Question question)  9  { 10         return async (id) =>
 { 12             if (new Random().Next(0, 2) == 0) 13  { 14                 Console.WriteLine("Exception Happened!!!!"); 15                 throw new Exception("Error Hanppened!"); 16  } 17             else
 { 19                 Console.WriteLine(string.Format("worker：{0}，content：{1}", id, question.Text)); 20  } 21  }; 22     }

订阅方法中声明了两个消息的订阅者（因为 subscriptionId相同，所以消息会采取轮询的方法，依次发送到每个消息的消费者）。

消息处理中产生随机数，进而有33%的机会产生异常

3.3消息发布

    var msg = Console.ReadLine();
    bus.Publish(new Question(msg));

发布程序很简单，读取输入的内容，直接使用EasyNetQ提供的发布方法即可。

3.4异常处理

 

    private static void HandleErrors()  2  {  3         Action<IMessage<Error>, MessageReceivedInfo> handleErrorMessage = HandleErrorMessage;  4 
        IQueue queue = new Queue(ErrorQueue, false);  6  bus.Advanced.Consume(queue, handleErrorMessage);  7  }  8 
    private static void HandleErrorMessage(IMessage<Error> msg, MessageReceivedInfo info) 10  { 11         Console.WriteLine("catch: " + msg.Body.Message); 12     }

 

异常处理程序订阅了队列“EasyNetQ_Default_Error_Queue”，当异常发生时，EasyNetQ默认的处理是将队列数据写入当前的错误队列中。

 

 

4.功能测试

 当程序异常时，打印异常

 

5.异常重试

实际项目中，当程序发生异常后，我们期望的处理可能是将消息返回到原有队列，进行再次的数据处理。

修改bus创建的声明：

bus = RabbitHutch.CreateBus("host=localhost", x => x.Register<IConsumerErrorStrategy>(_ => new AlwaysRequeueErrorStrategy()));

AlwaysRequeueErrorStrategy是我们默认的错误处理方法

    public sealed class AlwaysRequeueErrorStrategy : IConsumerErrorStrategy  2  {  3         public void Dispose()  4  {  5  }  6 
        public AckStrategy HandleConsumerError(ConsumerExecutionContext context, Exception exception)  8  {  9             return AckStrategies.NackWithRequeue; 10  } 11 
        public AckStrategy HandleConsumerCancelled(ConsumerExecutionContext context) 13  { 14             return AckStrategies.NackWithRequeue; 15  } 16     }