RabbitMQ学习系列四-EasyNetQ文档跟进式学习与实践

EasyNetQ文档跟进式学习与实践

https://www.cnblogs.com/DjlNet/p/7603554.html

这里可能有人要问了，为什么不使用官方的nuget包呐：RabbitMQ.Client（官方还在积极对.net core做升级去兼容.net standrad，这挺好，https://www.nuget.org/packages/RabbitMQ.Client/5.1.0-pre1），要说为什么，其实无非就是原始的官方包你说要用吧也可以用，就是需要学习成本，让小组成员都要熟悉API又是一番功夫且低级API方法不是那么通俗易懂，当然拉有能力的当然可以借助官方包二次封装以适应自己项目或者Team的需求，前提是对交互机制和对原始API有较为全面的了解，所以综上所说呐，在社区的支持下提供了 框架：EasyNetQ，从其名便可以知道，方便了我们使用高阶API的同时，依然保留了对原始高级功能点的访问，所以这也体现框架作者的对框架的理解也基本上足够我们当前的使用需求了，当然了此框架本身也是依赖于 RabbitMQ.Client 的也是一个二次封装的结果，所以呐博主选取它作为学习或者项目集成方面，都不愧是一个不错的选择！相信使用过的老铁，都知道它的好，希望作者延续出core的版本吧.....同时博主不希望只仅仅是简单的跟着文档滚一边，而且想以一种正确优雅的姿势把此消息队列框架集成在系统当中，且巧妙的设计以及避开一些坑，实现在业务与消息之间微妙的关系（例如通过事件总线来统筹，MQ消息来传递），同时保证消息交互一个不稳定的情况下，如何实现尽可能的消息落地与确保，这也是当今面对分布式事务的一种婉转但不失优雅的解决方案.....

EasyNetQ文档划重点与测试

这里是博主认为有必须要视为重点且需要实验验证的点，如若没有提及的地方，可以提出来一起讨论学习

1、发布者确认（Publisher Confirms）：简而言之就是确保消息的成功投递，通俗的讲就是我要知道，我投递的消息到底有木有成功被接受。连接字符增加 publisherConfirms=true; 即可，这里博主建议不要为了那么一丢丢的确认性能的考虑而放弃发送消息确认配置，同时为了保证消息的投递成功，RabbitMQ也推荐使用该方式因为没有实现事务方式，且发送消息之后不管是同步方法(Publish)还是异步方法(PublishAsync)在自身内部都会在超时还没到的时间段内，一直等待消息确认接收的反馈讯息，再然后同步方法才会返回，异步方法也会返回一个完成状态的Task对象，当然可能因为网络或者MQServer服务宕机等一系列问题导致同步异常爆发或者异步返回异常的Task对象，所以还需要对后续异常情况处理和日志记录机制的的实施，至此这种发布者确认带来的消息推送保障对于我们的系统来说很重要，关乎到系统交互的责任归属问题（例如：你说你投递成功了，我咋知道你投递成功了没有，我这里没收到啊，这样比较搞笑的对话了）

这里补充说明一点就是，日常开发中博主见过的消息推送目前的两种方式：

a、要求消息的发送与业务逻辑耦合在一个数据库的事务作用域当中，且消息发送需要放在业务逻辑的最后面（你懂的），这样就形成了一个利用数据库层面的强制原子操作，包含消息发送日志记录当作证据，达到消息和业务绑定关系为同生共死“要么都成功，要么都失败”，好处：实现了逻辑的一致性以及消息的及时发送，可以满足某些特殊需求，坏处：消息发送需要远程服务器交互（内网或者外网）需要等待耗时且一定程度拉低了系统处理能力受MQ消息服务器的影响，并且发送失败会导致业务处理失败浪费了前期逻辑处理，所以也不存在消息的重试机制了，直接重新操作业务逻辑就可以重试发送了；当然依然需要异常处理与日志记录

b、将业务逻辑产生的消息归档到一个等待发送消息DB（Table）中做持久化处理，然后通过定时器框架配合处理，轮询取出一定数量应该发送的消息（这里包括没发送成功需要重试的消息，一直怼）到MQServer中成功无异常后重置消息的发送标记位flag=true，重试消息的话RetryCount++，这里当然也是在一个DB事务当中且消息发送依然在尾部逻辑（你懂的），好处：消息发送与业务逻辑分离，职责划分明确增加了系统各自部分的稳定性不受消息服务影响，相互独立不受牵制，且已经具备了消息发送日志记录（前面说的等待发送消息持久化证明）作为证据，坏处：消息发送具有一定得延迟性也就是定时器的间隔时间，同理这方式依然需要异常处理与日志记录

当然以上两种方式都是基于额外数据库事务支持的情况下(加上PublisherConfirms共同保证消息落地性)，至于选择何种方式或者是说怎么样来灵活的组合使用，来应对系统消息发布的场景，看上面的解释也能大致有数了吧，其实也需要根据开发时的业务场景、网络、服务器资源等等综合考虑，不过一般情况博主推荐使用方式b即可，一般一般一般情况下对于消息的及时性要求不高，控制在一定的忍耐程度就行了，也是目前博主采取的方式来操作，不过得注意得时相关与网络传输交互都需要异常处理和日志记录来做问题得检查和恢复得支撑，至于上面两种方式的代码集成就比较明显了，毕竟思路明确了，博主这里打算后面框架集成部分给出参考吧。

2、prefetchcount：在RabbitMQServer获取Consumer消费者反馈的ack（确认消费标志）之前，RabbitMQ服务能够分发消息给消费者的最大值，连接字符串配置：prefetchcount=50;，默认是50，且这些消息处于内存队列中处于准备状态，从控制台界面查看就是(注意红框处就是设置值，这里博主测试设置了为:prefetchcount=5;，然后故意让消费者延迟反馈ack确认消息回执)

测试代码如下：

IBus bus = RabbitHutch.CreateBus("host=localhost;virtualHost=/;username=yourname;password=yourpassword;publisherConfirms=true;timeout=10;prefetchcount=5");
            bus.Subscribe<MyClass>(string.Empty, x =>
            {
                Console.WriteLine(x.Text);
                Task.Delay(TimeSpan.FromSeconds(5)).Wait();
            });

当然这个配置是作用于消费者，那么这里每个消费者可以自定义该配置来覆盖连接字符串中的全局配置，来达到控制个别消费者的细粒度，参考如下代码与效果图

，设置0表示不限制可以无限发送消息（不推荐），设置1表示消费者以公平方式接受消息(多个消费者共享一个队列的时候，消费顺序按照订阅而定)，其实说白了，这个值的设定是要看消息接收方处理能力而定的，所以具体生成环境看情况而定吧，采用默认设置也可以。

3、Ack：表示订阅者接受并处理消息之后回执。

例如：一个订阅了MyMessage的订阅者，如果程序崩溃会发生什么？为了提高效率，EasyNetQ实现了一个用于订阅的内部内存队列。消息通过网络从RabbitMQ接收并放置在此队列上。单个订阅线程依次将消息从队列中取出，并将它们传递给您提供的回调。一旦回调完成，EasyNetQ将发送一个“Ack”回RabbitMQ。在收到“Ack”之前，消息不会从RabbitMQ队列中删除。如果您的服务在处理消息时中止，还没有Ack的回传，消息（以及EasyNetQ内存中队列中的所有消息）将保留在RabbitMQ队列中，一旦您的服务重新连接，消息将被重新发送，从上而知也知道ACK的作用和意义。

当然消费消息的时候发生异常：如果您的订阅回调引发EasyNetQException异常，EasyNetQ将会收到正在消费的消息，并将其包装在特殊的错误消息中。错误消息将被发布到EasyNetQ错误队列（名为EasyNetQ_Default_Error_Queue）。您应该监视任何消息的错误队列。错误消息包括重新发布原始消息以及异常类型，消息和堆栈跟踪所需的所有信息。您可以使用EasyNetQ.Hosepipe实用程序重新发布错误消息。请参阅下面的EasyNetQ.Hosepipe部分。

这里默认EasyNetQ已经帮我们做好了消息处理的部分逻辑，包括异常处理和消息ACK回执，

那么我们在生产环境如何保证消费消息做到尽可能消息100%落地？

这里博主的大致解决办法描述如下：这里消息体的格式协定以及消息基本验证什么的，不再讨论方法之类，就如同API方法参考校验一个性质，这里相信大部分订阅者处理消息都参与了数据库逻辑，所以这里博主就必须把传递过来的消息持久化到数据库中当做消费证明，当然这里又可以使用数据库的ACID的事务机制来包裹消息日记录ConsumerLog以及之前的一些业务逻辑数据库写操作或者其他Web请求逻辑，当然可能会由于网络请求或者数据库本身的问题出现爆发异常，导致事务回滚（业务逻辑同时消费消息日志记录失败，这种就是除本身逻辑异常之外的不可控的异常爆发），那么这种还需要进一步保证消息落地怎么办呐？回答：没有百分百的保证，首先还是再一次的向数据库或者NoSql再次写入消息，不过这里是写入异常消费失败的消息ConsumerErrorLog（那么就需要定时器消息重试消费逻辑的机制），如果上述还是不行（因为这里还可能还是失败）就需要文本日志来记录Message消息体做进一步保证了，其次就是Log记录异常原因，最最后我还可以利用上述EasyNetQ提供了EasyNetQException异常机制将消费失败的消息进一步在EasyNetQ错误队列（名为EasyNetQ_Default_Error_Queue）也保留一份存档方便对照查阅，到这里相信基本上能够尽可能保证消息消费的正确姿势了，当然上述思路可能依然并不是最合适的，最好是结合自己的情况进一步定制化自己的消息处理框架吧。