消息队列小结

消息队列优势劣势

消息队列最常被使用的三种场景:异步处理、流量控制和服务解耦。当然,消息队列的适用范围不仅仅局限于这些场景,还有包括:

  • 作为发布/订阅系统实现一个微服务级系统间的观察者模式;
  • 连接流计算任务和数据;
  • 用于将消息广播给大量接收者。

同时我们也要认识到,消息队列也有它自身的一些问题和局限性,包括:

  • 引入消息队列带来的延迟问题;
  • 增加了系统的复杂度;
  • 可能产生数据不一致的问题。

该如何选择消息队列?

如果说,消息队列并不是你将要构建系统的主角之一,你对消息队列功能和性能都没有很高的要求,只需要一个开箱即用易于维护的产品,我建议你使用RabbitMQ。

如果你的系统使用消息队列主要场景是处理在线业务,比如在交易系统中用消息队列传递订单,那RocketMQ的低延迟和金融级的稳定性是你需要的。

如果你需要处理海量的消息,像收集日志、监控信息或是前端的埋点这类数据,或是你的应用场景大量使用了大数据、流计算相关的开源产品,那Kafka是最适合你的消息队列。

第二梯队的消息队列

先说ActiveMQ,ActiveMQ是最老牌的开源消息队列,是十年前唯一可供选择的开源消息队列,目前已进入老年期,社区不活跃。无论是功能还是性能方面,ActiveMQ都与现代的消息队列存在明显的差距,它存在的意义仅限于兼容那些还在用的爷爷辈儿的系统。

接下来说说ZeroMQ,严格来说ZeroMQ并不能称之为一个消息队列,而是一个基于消息队列的多线程网络库,如果你的需求是将消息队列的功能集成到你的系统进程中,可以考虑使用ZeroMQ。

最后说一下Pulsar,很多人可能都没听说过这个产品,Pulsar是一个新兴的开源消息队列产品,最早是由Yahoo开发,目前处于成长期,流行度和成熟度相对没有那么高。与其他消息队列最大的不同是,Pulsar采用存储和计算分离的设计,它有可能会引领未来消息队列的一个发展方向,建议你持续关注这个项目。

avatar

引入mq会多哪些问题

重复消息问题

重复消费问题可以说是mq中普遍存在的问题,不管你用哪种mq都无法避免。有哪些场景会出现重复的消息呢?

  • 消息生产者产生了重复的消息
  • kafka和rocketmq的offset被回调了
  • 消息消费者确认失败
  • 消息消费者确认时超时了
  • 业务系统主动发起重试

数据一致性问题

很多时候,如果mq的消费者业务处理异常的话,就会出现数据一致性问题。比如:一个完整的业务流程是,下单成功之后,送100个积分。下单写库了,但是消息消费者在送积分的时候失败了,就会造成数据不一致的情况,即该业务流程的部分数据写库了,另外一部分没有写库。

如果下单和送积分在同一个事务中,要么同时成功,要么同时失败,是不会出现数据一致性问题的。

但由于跨系统调用,为了性能考虑,一般不会使用强一致性的方案,而改成达成最终一致性即可。

消息丢失问题

同样消息丢失问题,也是mq中普遍存在的问题,不管你用哪种mq都无法避免。有哪些场景会出现消息丢失问题呢?

  • 消息生产者发生消息时,由于网络原因,发生到mq失败了。
  • mq服务器持久化时,磁盘出现异常
  • kafka和rocketmq的offset被回调时,略过了很多消息。
  • 消息消费者刚读取消息,已经ack确认了,但业务还没处理完,服务就被重启了。

导致消息丢失问题的原因挺多的,生产者、mq服务器、消费者 都有可能产生问题,我在这里就不一一列举了。最终的结果会导致消费者无法正确的处理消息,而导致数据不一致的情况。

消息顺序问题

有些业务数据是有状态的,比如订单有:下单、支付、完成、退货等状态,如果订单数据作为消息体,就会涉及顺序问题了。如果消费者收到同一个订单的两条消息,第一条消息的状态是下单,第二条消息的状态是支付,这是没问题的。但如果第一条消息的状态是支付,第二条消息的状态是下单就会有问题了,没有下单就先支付了?

消息顺序问题是一个非常棘手的问题,比如:

  • kafka同一个partition中能保证顺序,但是不同的partition无法保证顺序。
  • rabbitmq的同一个queue能够保证顺序,但是如果多个消费者同一个queue也会有顺序问题。

如果消费者使用多线程消费消息,也无法保证顺序。

如果消费消息时同一个订单的多条消息中,中间的一条消息出现异常情况,顺序将会被打乱。

还有如果生产者发送到mq中的路由规则,跟消费者不一样,也无法保证顺序。

消息堆积

如果消息消费者读取消息的速度,能够跟上消息生产者的节奏,那么整套mq机制就能发挥最大作用。但是很多时候,由于某些批处理,或者其他原因,导致消息消费的速度小于生产的速度。这样会直接导致消息堆积问题,从而影响业务功能。

解决问题方案

重复消息问题

增加一张消费消息表,来解决mq的这类问题。消费消息表中,使用messageId做唯一索引,在处理业务逻辑之前,先根据messageId查询一下该消息有没有处理过,如果已经处理过了则直接返回成功,如果没有处理过,则继续做业务处理。

avatar

数据一致性问题

数据一致性分为:强一致性、弱一致性、最终一致性

而mq为了性能考虑使用的是最终一致性,那么必定会出现数据不一致的问题。这类问题大概率是因为消费者读取消息后,业务逻辑处理失败导致的,这时候可以增加重试机制。

重试分为:同步重试 和 异步重试。

有些消息量比较小的业务场景,可以采用同步重试,在消费消息时如果处理失败,立刻重试3-5次,如何还是失败,则写入到记录表中。但如果消息量比较大,则不建议使用这种方式,因为如果出现网络异常,可能会导致大量的消息不断重试,影响消息读取速度,造成消息堆积。

avatar

而消息量比较大的业务场景,建议采用异步重试,在消费者处理失败之后,立刻写入重试表,有个job专门定时重试。

还有一种做法是,如果消费失败,自己给同一个topic发一条消息,在后面的某个时间点,自己又会消费到那条消息,起到了重试的效果。如果对消息顺序要求不高的场景,可以使用这种方式。

消息丢失问题

不管你是否承认有时候消息真的会丢,即使这种概率非常小,也会对业务有影响。生产者、mq服务器、消费者都有可能会导致消息丢失的问题。

为了解决这个问题,我们可以增加一张消息发送表,当生产者发完消息之后,会往该表中写入一条数据,状态status标记为待确认。如果消费者读取消息之后,调用生产者的api更新该消息的status为已确认。有个job,每隔一段时间检查一次消息发送表,如果5分钟(这个时间可以根据实际情况来定)后还有状态是待确认的消息,则认为该消息已经丢失了,重新发条消息。

avatar

消息顺序问题

消息顺序问题是我们非常常见的问题,我们以kafka消费订单消息为例。订单有:下单、支付、完成、退货等状态,这些状态是有先后顺序的,如果顺序错了会导致业务异常。

解决这类问题之前,我们先确认一下,消费者是否真的需要知道中间状态,只知道最终状态行不行?

这种方式可以解决大部分的消息顺序问题。

但如果真的有需要保证消息顺序的需求。订单号路由到不同的partition,同一个订单号的消息,每次到发到同一个partition。

avatar

消息堆积

那么消息堆积问题该如何解决呢?

这个要看消息是否需要保证顺序。

如果不需要保证顺序,可以读取消息之后用多线程处理业务逻辑。

avatar

这样就能增加业务逻辑处理速度,解决消息堆积问题。但是线程池的核心线程数和最大线程数需要合理配置,不然可能会浪费系统资源。

如果需要保证顺序,可以读取消息之后,将消息按照一定的规则分发到多个队列中,然后在队列中用单线程处理

avatar

半事务消息

事务消息就是MQ提供的类似XA的分布式事务能⼒,通过事务消息可以达到分布式事务的最终⼀致性。

半事务消息就是MQ收到了⽣产者的消息,但是没有收到⼆次确认,不能投递的消息。

实现原理如下:

  1. ⽣产者先发送⼀条半事务消息到MQ
  2. MQ收到消息后返回ack确认
  3. ⽣产者开始执⾏本地事务
  4. 如果事务执⾏成功发送commit到MQ,失败发送rollback
  5. 如果MQ⻓时间未收到⽣产者的⼆次确认commit或者rollback,MQ对⽣产者发起消息回查
  6. ⽣产者查询事务执⾏最终状态
  7. 根据查询事务状态再次提交⼆次确认

最终,如果MQ收到⼆次确认commit,就可以把消息投递给消费者,反之如果是rollback,消息会保存下来并且在3天后被删除。

avatar

参考文献

  1. https://mp.weixin.qq.com/s/9tUnhdVRHHkpGpYUQdFVuQ