NetCore EasyNetQ 高级使用 RabbitMq_坚定的走-CSDN博客

来源： NetCore EasyNetQ 高级使用 RabbitMq_坚定的走-CSDN博客

一、消息队列

消息队列作为分布式系统中的重要组件，常用的有MSMQ，RabbitMq，Kafa，ActiveMQ，RocketMQ。至于各种消息队列的优缺点比较，在这里就不做扩展了，网上资源很多。

更多内容可参考 消息队列及常见消息队列介绍。我在这里选用的是RabbitMq。

官网地址：http://www.rabbitmq.com

安装和配置：Windows下RabbitMq安装及配置

二、RabbitMq简单介绍

RabbitMQ是一款基于AMQP（高级消息队列协议），由Erlang开发的开源消息队列组件。是一款优秀的消息队列组件，他由两部分组成：服务端和客户端，客户端支持多种语言的驱动，如：.Net、JAVA、   Erlang等。在RabbitMq中首先要弄清楚的概念是 交换机、队列、绑定。基本的消息通讯步骤就是首先定义ExChange，然后定义队列，然后绑定交换机和队列。

需要明确的一点儿是，发布者在发送消息是，并不是把消息直接发送到队列中，而是发送到Exchang，然后由交互机根据定义的消息匹配规则，在将消息发送到队列中。

Exchange有四种消息消息分发规则：direct，topic，fanout，header。headers 匹配 AMQP 消息的 header 而不是路由键，此外 headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了。

详细的概念介绍推荐查看：消息队列之RabbitMq

三、EasyNetQ使用

Easynetq是一个简单易用的Rabbitmq Net客户端。同时支持 NetFramework和NetCore。GitHub地址。它是针对RabbitMq Net客户端的进一步封装。关于EasyNetQ的简单使用推荐教程：EasyNetQ的介绍。

本文主要介绍基于EasyNeq的高级API的使用。EasyNetQ的作者在核心的IBus接口中尽量避免暴露AMQP中的交换机、队列、绑定这些概念，使用者即使不去了解这些概念，也能完成消息的发送接收。这相当简洁，但某些情况下，基于应用场景的需要，我们需要自定义交换机、队列、绑定这些信息，EasyNetQ允许你这么做，这些都是通过IAdvanceBus接口实现。

3.1 项目装备

这里为了演示，首先新建一个项目，包括一个发布者，两个接收者，一个公共的类库

安装EasyNetQ： NuGet>Install-Package EasyNetQ

3.2 简单封装

在Common项目里面是针对Easynetq的使用封装，主要目录如下

在RabbitMq文件夹下，是针对消息发送接收的简单封装。

首先来看下RabbitMqManage，主要的发送和订阅操作都在这个类中。其中ISend接口定义了发送消息的规范，SendMessageManage是ISend的实现。IMessageConsume接口定义订阅规范。

MesArg 和PushMsg分别是订阅和发送需用到的参数类。RabbitMQManage是暴露在外的操作类。

首先看发送的代码

1. public enum SendEnum
2. {
3. 订阅模式 = 1,
4. 推送模式 = 2,
5. 主题路由模式 = 3
6. }
7. public class PushMsg
8. {
9. /// <summary>
10. /// 发送的数据
11. /// </summary>
12. public object sendMsg { get; set; }
14. /// <summary>
15. /// 消息推送的模式
16. /// 现在支持：订阅模式,推送模式,主题路由模式
17. /// </summary>
18. public SendEnum sendEnum { get; set; }
20. /// <summary>
21. /// 管道名称
22. /// </summary>
23. public string exchangeName { get; set; }
25. /// <summary>
26. /// 路由名称
27. /// </summary>
28. public string routeName { get; set; }
29. }
32. internal interface ISend
33. {
34. Task SendMsgAsync(PushMsg pushMsg, IBus bus);
36. void SendMsg(PushMsg pushMsg, IBus bus);
37. }
39. internal class SendMessageMange : ISend
40. {
41. public async Task SendMsgAsync(PushMsg pushMsg, IBus bus)
42. {
43. //一对一推送
45. var message = new Message<object>(pushMsg.sendMsg);
46. IExchange ex = null;
47. //判断推送模式
48. if (pushMsg.sendEnum == SendEnum.推送模式)
49. {
50. ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Direct);
51. }
52. if (pushMsg.sendEnum == SendEnum.订阅模式)
53. {
54. //广播订阅模式
55. ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Fanout);
56. }
57. if (pushMsg.sendEnum == SendEnum.主题路由模式)
58. {
59. //主题路由模式
60. ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Topic);
61. }
62. await bus.Advanced.PublishAsync(ex, pushMsg.routeName.ToSafeString(“”), false, message)
63. .ContinueWith(task =>
64. {
65. if (!task.IsCompleted && task.IsFaulted)//消息投递失败
66. {
67. //记录投递失败的消息信息
68. }
69. });
72. }
74. public void SendMsg(PushMsg pushMsg, IBus bus)
75. {
76. //一对一推送
78. var message = new Message<object>(pushMsg.sendMsg);
79. IExchange ex = null;
80. //判断推送模式
81. if (pushMsg.sendEnum == SendEnum.推送模式)
82. {
83. ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Direct);
84. }
85. if (pushMsg.sendEnum == SendEnum.订阅模式)
86. {
87. //广播订阅模式
88. ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Fanout);
89. }
90. if (pushMsg.sendEnum == SendEnum.主题路由模式)
91. {
92. //主题路由模式
93. ex = bus.Advanced.ExchangeDeclare(pushMsg.exchangeName, ExchangeType.Topic);
94. }
95. bus.Advanced.Publish(ex, pushMsg.routeName.ToSafeString(“”), false, message);
98. }
100. }

在EasyNetQ中对于异步发送消息的时候，消息是否送达Broker只需要查看异步发送方法最终执行成功还是失败，成功就表示消息送达，如果失败可以将失败后的消息存入数据库中，然后用后台线程轮询

数据库表，将失败后的消息进行重新 发送。这种方式还可以进一步变成消息表，就是先将要发送的消息存入消息表中，然后后台线程轮询消息表来进行消息发送。一般这种方式被广泛用于分布式事务中，

将本地数据库操作和消息表写入放入同一个本地事务中，来保证消息发送和本地数据操作的同步成功，因为我的系统中，分布式事务的涉及很少，所以就没这样去做，只是简单的在异步发送的时候监控下

是否发送失败，然后针对失败的消息做一个重新发送的机制。这里，推荐大佬的NetCore分布式事务解决方案 CAP GitHub地址。

 接着看一下消息订阅接收涉及的代码

1. public class MesArgs
2. {
3. /// <summary>
4. /// 消息推送的模式
5. /// 现在支持：订阅模式,推送模式,主题路由模式
6. /// </summary>
7. public SendEnum sendEnum { get; set; }
9. /// <summary>
10. /// 管道名称
11. /// </summary>
12. public string exchangeName { get; set; }
14. /// <summary>
15. /// 对列名称
16. /// </summary>
17. public string rabbitQueeName { get; set; }
19. /// <summary>
20. /// 路由名称
21. /// </summary>
22. public string routeName { get; set; }
23. }
24. public interface IMessageConsume
25. {
26. void Consume(string message);
27. }

在订阅中我定义了一个接口，最终业务代码中，所有的消息订阅类，都需要继续此接口

最后，我们来看下对外使用的操作类

1. public class RabbitMQManage
2. {
4. private volatile static IBus bus = null;
6. private static readonly object lockHelper = new object();
8. /// <summary>
9. /// 创建服务总线
10. /// </summary>
11. /// <param name=”config”></param>
12. /// <returns></returns>
13. public static IBus CreateEventBus()
14. {
15. //获取RabbitMq的连接地址
16. //SystemJsonConfigManage 是我简单封装的一个json操作类，用于针对json文件的读写操作
17. var config = SystemJsonConfigManage.GetInstance().AppSettings[“MeessageService”];
18. if (string.IsNullOrEmpty(config))
19. throw new Exception(“消息地址未配置”);
20. if (bus == null && !string.IsNullOrEmpty(config))
21. {
22. lock (lockHelper)
23. {
24. if (bus == null)
25. bus = RabbitHutch.CreateBus(config);
26. }
27. }
28. return bus;
29. }
30. /// <summary>
31. /// 释放服务总线
32. /// </summary>
33. public static void DisposeBus()
34. {
35. bus?.Dispose();
36. }
37. /// <summary>
38. /// 消息同步投递
39. /// </summary>
40. /// <param name=”pushMsg”></param>
41. /// <returns></returns>
42. public static bool PushMessage(PushMsg pushMsg)
43. {
44. bool b = true;
45. try
46. {
47. if (bus == null)
48. CreateEventBus();
49. new SendMessageMange().SendMsg(pushMsg, bus);
50. b = true;
51. }
52. catch (Exception ex)
53. {
55. b = false;
56. }
57. return b;
58. }
59. /// <summary>
60. /// 消息异步投递
61. /// </summary>
62. /// <param name=”pushMsg”></param>
63. public static async Task PushMessageAsync(PushMsg pushMsg)
64. {
65. try
66. {
67. if (bus == null)
68. CreateEventBus();
69. await new SendMessageMange().SendMsgAsync(pushMsg, bus);
70. }
71. catch (Exception ex)
72. {
73. throw ex;
74. }
75. }
76. /// <summary>
77. /// 消息订阅
78. /// </summary>
79. public static void Subscribe<TConsum>(MesArgs args)
80. where TConsum : IMessageConsume,new()
81. {
82. if (bus == null)
83. CreateEventBus();
84. if (string.IsNullOrEmpty(args.exchangeName))
85. return;
86. Expression<Action<TConsum>> methodCall;
87. IExchange ex = null;
88. //判断推送模式
89. if (args.sendEnum == SendEnum.推送模式)
90. {
91. ex = bus.Advanced.ExchangeDeclare(args.exchangeName, ExchangeType.Direct);
93. }
94. if (args.sendEnum == SendEnum.订阅模式)
95. {
96. //广播订阅模式
97. ex = bus.Advanced.ExchangeDeclare(args.exchangeName, ExchangeType.Fanout);
98. }
99. if (args.sendEnum == SendEnum.主题路由模式)
100. {
101. //主题路由模式
102. ex = bus.Advanced.ExchangeDeclare(args.exchangeName, ExchangeType.Topic);
103. }
104. IQueue qu;
105. if (string.IsNullOrEmpty(args.rabbitQueeName))
106. {
107. qu = bus.Advanced.QueueDeclare();
108. }
109. else
110. qu = bus.Advanced.QueueDeclare(args.rabbitQueeName);
111. bus.Advanced.Bind(ex, qu, args.routeName.ToSafeString(“”));
112. bus.Advanced.Consume(qu, (body, properties, info) => Task.Factory.StartNew(() =>
113. {
114. try
115. {
116. lock (lockHelper)
117. {
119. var message = Encoding.UTF8.GetString(body);
120. //处理消息
121. methodCall = job => job.Consume(message);
122. methodCall.Compile()(new TConsum());
123. }
125. }
126. catch (Exception e)
127. {
129. throw e;
131. }
134. }));
135. }
136. }

这里面主要封装了消息的发送和订阅，以及IBus单例的创建。在后续的消息发送和订阅主要就通过此处来实现。我们看到一开始的类目结构中还有一个RaExMessageHandleJob类，这个类就是一个后台

循环任务，用来监测数据库中是否保存了发送失败的消息，如果有，则将消息取出，尝试重新发送。在此就不做多的介绍，大家可以根据自己的实际需求来实现。

3.3 发布者

现在来看一下消息发布者的代码

主要的发送代码都在Send类中，其中appsettings.json里面配置了Rabbitmq的连接地址，TestDto只是一个为了方便演示的参数类。

下面看一下Program里面的代码

很简单的一个发送消息调用。

然后来看一下Send类中的代码

1. public class Send
2. {
3. /// <summary>
4. /// 发送消息
5. /// </summary>
6. public static void SendMessage()
7. {
8. //需要注意一点儿，如果发送的时候，在该管道下找不到相匹配的队列框架将默认丢弃该消息
13. //推送模式
14. //推送模式下，需指定管道名称和路由键值名称
15. //消息只会被发送到和指定路由键值完全匹配的队列中
16. var directdto = new PushMsg()
17. {
18. sendMsg = new TestDto()
19. {
20. Var1 = “这是推送模式”
21. },
22. exchangeName = “message.directdemo”,
23. routeName= “routekey”,
24. sendEnum =SendEnum.推送模式
25. };
26. //同步发送 ，返回true或fasle true 发送成功，消息已存储到Rabbitmq中，false表示发送失败
27. var b= RabbitMQManage.PushMessage(directdto);
28. //异步发送，如果失败，失败的消息会被写入数据库，会有后台线程轮询数据库进行重新发送
29. //RabbitMQManage.PushMessageAsync(directlist);
32. //订阅模式
33. //订阅模式只需要指定管道名称
34. //消息会被发送到该管道下的所有队列中
35. var fanoutdto = new PushMsg()
36. {
37. sendMsg = new TestDto()
38. {
39. Var1 = “这是订阅模式”
40. },
41. exchangeName = “message.fanoutdemo”,
42. sendEnum = SendEnum.订阅模式
43. };
44. //同步发送
45. var fb = RabbitMQManage.PushMessage(fanoutdto);
46. //异步发送
47. //RabbitMQManage.PushMessageAsync(fanoutdto);
49. //主题路由模式
50. //路由模式下需指定 管道名称和路由值
51. //消息会被发送到该管道下，和路由值匹配的队列中去
52. var routedto = new PushMsg()
53. {
54. sendMsg = new TestDto()
55. {
56. Var1 = “这是主题路由模式1”,
57. },
58. exchangeName = “message.topicdemo”,
59. routeName=“a.log”,
60. sendEnum=SendEnum.主题路由模式
62. };
63. var routedto2 = new PushMsg()
64. {
65. sendMsg = new TestDto()
66. {
67. Var1 = “这是主题路由模式2”,
68. },
69. exchangeName = “message.topicdemo”,
70. routeName = “a.log.a.b”,
71. sendEnum = SendEnum.主题路由模式
73. };
74. //同步发送
75. var rb = RabbitMQManage.PushMessage(routedto);
76. var rb2 = RabbitMQManage.PushMessage(routedto2);
77. //异步发送
78. //RabbitMQManage.PushMessageAsync(routedto);
79. }
80. }

3.4 消费者

首先来看下消费者端的目录结构

其中appsettings.json中配置Rabbitmq的连接信息，Program中只是简单调用消息订阅

主要的消息订阅代码都在MessageManage文件夹下，MessageManService用于定义消息订阅类型

1. public class MessageManService
2. {
3. public static void Subsribe()
4. {
5. Task.Run(() =>
6. {
7. //概念 一个管道下面可以绑定多个队列。
8. //发送消息 是指将消息发送到管道中，然后由rabbitmq根据发送规则在将消息具体的转发到对应到管道下面的队列中
9. //消费消息 是指消费者（即服务）从管道下面的队列中获取消息
10. //同一个队列 可以有多个消费者（即不同的服务，都可以连接到同一个队列去获取消息）
11. //但注意 当一个队列有多个消费者的时候，消息会被依次分发到不同的消费者中。比如第一条消息给第一个消费者，第二条消息给第二个消费者(框架内部有一个公平分发的机制)
14. //推送模式时 需指定管道名称和路由值
15. //队列名称可自己指定
16. //注意 ，管道名称和路由名称一定要和发送方的管道名称和路由名称一致
17. //无论这个管道下面挂靠有多少个队列，只有路由名称和此处指定的路由名称完全一致的队列，才会收到这条消息。
18. var dirarg = new MesArgs()
19. {
20. sendEnum = SendEnum.推送模式,
21. exchangeName = “message.directdemo”,
22. rabbitQueeName = “meesage.directmessagequene”,
23. routeName = “routekey”
24. };
25. RabbitMQManage.Subscribe<DirectMessageConsume>(dirarg);
27. //订阅模式时需指定管道名称，并且管道名称要和发送方管道名称一致
28. //队列名称可自己指定
29. //所有这个管道下面的队列，都将收到该条消息
30. var fanoutrg = new MesArgs()
31. {
32. sendEnum = SendEnum.订阅模式,
33. exchangeName = “message.fanoutdemo”,
34. rabbitQueeName = “meesage.fanoutmessagequene”
35. };
36. RabbitMQManage.Subscribe<FanoutMessageConsume>(fanoutrg);
38. //路由模式时需指定管道名称，路由关键字并且管道名称，路由关键字要和发送方的一致
39. //队列名称可自己指定
40. //消息将被发送到管道下面的能匹配路由关键字的队列中
41. //也就是说 路由模式时，有多少队列能收到消息，取决于该队列的路由关键字是否匹配，只要匹配就能收到消息
42. //符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词
43. var topicrg = new MesArgs()
44. {
45. sendEnum = SendEnum.主题路由模式,
46. exchangeName = “message.topicdemo”,
47. rabbitQueeName = “message.topicmessagequene”,
48. routeName = “#.log.#”
49. };
51. RabbitMQManage.Subscribe<TopicMessageConsume>(topicrg);
52. });
53. }
54. }

Consume文件夹下主要定义了消息的业务处理

1. //推送模式过来的消息
2. public class DirectMessageConsume : IMessageConsume
3. {
4. //消息的处理方法中最好不要进行try catch操作
5. //如果发送异常，EasyNetQ会自动将消息放入错误队列中
6. //如果在Consume方法体中捕获了异常并且没有抛出，会默认消息处理成功
7. //消息的幂等性需业务方自行处理,也就是说同一条消息可能会接收到两次
8. //（比如说第一次正在处理消息的时候服务挂掉，服务重启后这条消息又会重新推送过来）
9. public void Consume(string message)
10. {
11. var dto = JsonConvert.DeserializeObject<TestDto>(message);
12. Console.WriteLine(dto.Var1 + “;” + dto.Var2 + “;” + dto.Var3);
13. }
14. }
16. //广播模式过来的消息
17. public class FanoutMessageConsume : IMessageConsume
18. {
19. //消息的处理方法中最好不要进行try catch操作
20. //如果发送异常，EasyNetQ会自动将消息放入错误队列中
21. //如果在Consume方法体中捕获了异常并且没有抛出，会默认消息处理成功
22. //消息的幂等性需业务方自行处理,也就是说同一条消息可能会接收到两次
23. //（比如说第一次正在处理消息的时候服务挂掉，服务重启后这条消息又会重新推送过来）
24. public void Consume(string message)
25. {
26. var dto = JsonConvert.DeserializeObject<TestDto>(message);
27. Console.WriteLine(dto.Var1 + “;” + dto.Var2 + “;” + dto.Var3);
28. }
29. }
31. //主题路由模式过来的消息
32. public class TopicMessageConsume : IMessageConsume
33. {
34. //消息的处理方法中最好不要进行try catch操作
35. //如果发送异常，EasyNetQ会自动将消息放入错误队列中
36. //如果在Consume方法体中捕获了异常并且没有抛出，会默认消息处理成功
37. //消息的幂等性需业务方自行处理,也就是说同一条消息可能会接收到两次
38. //（比如说第一次正在处理消息的时候服务挂掉，服务重启后这条消息又会重新推送过来）
39. public void Consume(string message)
40. {
41. var dto = JsonConvert.DeserializeObject<TestDto>(message);
42. Console.WriteLine(dto.Var1 + “;” + dto.Var2 + “;” + dto.Var3);
43. }
44. }

可以看到，所有的类都集成自我们定义的接口IMessageConsume。

四、总结

在EasyNetQ中如果需要消费者确认功能，则需要在Rabbitmq的连接配置中设置publisherConfirms=true，这将会开启自动确认。在使用高级api定义交换机和队列时可以自己定义多种参数，比如消息是否持久化，消息最大长度等等，具体大家可以去看官方文档，上面有详细介绍。Easynetq会自动去捕获消费异常的消息并将其放入到错误队列中，而且官方提供了重新发送错误队列中消息的方法，当然你也可以自己去监视错误列队，对异常消息进行处理。EasyNetQ里面作者针对消息的发布确认和消费确认都做了封装。在EasyNetQ中发布消息的时候如果选用的同步发送，只要没有抛出异常，我们就可以认为任务消息已经正确到达Broker，而异步发送的话需要我们自己去监视Task是否成功 。如果开启了自动确认，并不需要我们在消息处理的方法体中手动返回ack信息，只要消息被 正确处理就会自动ack。虽然RabbitMq中也有事务消息，但由于性能比较差，并不推荐使用。其实，只要我们能明确消息是否发布成功和消费成功，就将会很容易在这个基础上扩展出分布式事务的处理。