我要评论为什么要使用rabbitmq
rabbitmq是一款非常优秀的消息中间件,它可以实现各个模块之间的松耦合,提高程序的灵活性,可扩展性,使用rabbitmq主要有以下优点:
- 异步消息传递,rabbitmq支持异步消息传递,可以实现异步处理消息,提高程序的执行效率.
- 消息队列排队服务,rabbitmq可以将消息暂存到消息队列中,有效缓解生产者和消费者的压力.
- rabbitmq支持基于消息的路由机制,发布订阅/机制,工作模式等多重模式,灵活性较高.
- rabbitmq非常容易扩展,对于大数据量或者高并发的程序的帮助非常大.
- 可靠性,rabbitmq有非常可靠的消息确认机制,如:ack确认机制和消息的持久性等.确保消息不会丢失..
说一下你在项目中使用rabbitmq的场景
我们的项目使用rabbitmq的场景非常多,我也在很多场景使用过rabbitmq,例如:
在用户选择调整上课时间之后,会从原直播间的白名删除,再加入到新的直播间白名单中,并且通知对应的分管教师.在这个流程中,最核心的业务功能是从原直播间的白名单中删除,再加入到新的直播间白名单中.其次不是很重要的是通知对应的分管教师.在这个场景下,我们同步完成的业务就是刚才提到的核心业务.在核心业务处理完之后,会从知音楼、教师后台、短信同时通知对应的分管教师.这三个通知我们会使用rabbitmq进行异步处理,这样可以提高整个业务逻辑功能的执行速度,提高用户体验.这个场景主要也是使用了rabbitmq的异步处理功能.
还是上面的例子.由于我们的项目是分布式的项目,各个模块之间是相互独立的,从接受到请求的controller层到service层再到dao层,其实都是一些独立的子项目,在子项目中相互交流,我们会使用rabbitmq进行消息传递,这样既可以降低各个模块之间的耦合度,也可以确保消息不会丢失.
而且由于我们的项目要求所有的操作日志都要存放在相应的日志表里,并且同时要落盘到本地的日志文件中,所以为了提高效率,我们会将日志信息存放到rabbitmq中,由专门负责处理日志的模块去从消息队列中获取消息并处理.这样不会因为处理日志信息导致核心业务的处理效率收到日志处理的影响.
如何确保消息正确的发送到rabbitmq?如何确保消息接收方接收到了消息
在使用rabbitmq时,我们要首先确认程序和rabbitmq的连接状态和信道状态
之后就要确保生产者的消息可以到达exchange,这一环节rabbitmq是提供了ack消息确认机制的,当我们开启ack消息确认机制.消息从生产者发送到交换机,会有一个confirm回调函数,如果交换机正确的接收了方法,ack为true,如果交换机没有接收到消息,ack就是false,同时还会返回对应的错误信息.
要确保消息接收方能够接收到消息,我们要做一下分析,使用rabbitmq之后,生产者会将消息先发送至exchange,exchange再将消息发送到相应的queue,消费者从queue中获取消息，其中第一个环节我们已经可以确保它可以发送过去。最后的消费者从queue获取消息，我们需要检查消费者监听的队列是否正确。
重点在第二步，从exchange到queue，这个环节rabbitmq提供了return确认机制，如果消息从exchange成功到达了queue，不会执行回调，如果发送失败，就会执行returncallback回调，同时返回消息本体，错误状态码，错误信息，交换机和发生错误的路由。
如何确认消息不会被重复消费
按照正常逻辑来说,消费者在消费一条消息之后会向mq发送一条消息,mq会在收到消息之后删除该消息.但是可能由于一些原因,消费者在获取到消息之后在给mq发送消息时宕机之类的就会导致消息被消费但是没有从队列中移除.再次被消费时就会导致消息重复消费的问题.
解决方式
我们消息的生产者在发送消息时,给每条消息添加一个唯一的消息id,同时在redis中以id为key添加一条针对消费状态的数据,在消费者收到消息先根据消息id去redis中查询是否被消费了,如果消费的话后面的业务逻辑就不再执行.如果没有消费就继续执行后面的业务.在执行完业务之后将该条消息在redis中的状态修改为已消费.
这样的话如果当前的消息被消费并且业务执行完成了,但是队列中没有移除消息.当消费者再次消费当前消息时,还会去redis中查询消费状态,如果,被消费了,就不去执行后面的业务,这样就不会重复消费.虽然消息还是被消费了两次,但是对业务不会造成影响.
发表评论