EasyNetQ异常处理

Posted 2022-01-29 imstrive

代码下载

https://github.com/longtaosu/DotnetCoreDemo/tree/master/01RabbitMQ/02EasyNetQ/9.%E8%A1%A5%E5%85%85Demo/9.1%E5%BC%82%E5%B8%B8%E5%A4%84%E7%90%86/EasyNetQDemo.ErrorHandler

官方Demo

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

 

1.新建项目

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

 

2.创建消息模型

2.1创建Answer

创建Answer数据模型

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

2.2创建Question

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

 

3.测试程序

3.1初始化代码

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

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

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

 

3.2消息订阅

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

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

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

3.3消息发布

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

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