用MongoDB取代RabbitMQ

RabbitMQ是当成应用比较广泛的队列服务系统,其配套的客户端和监控运维方案也比较成熟。BoxedIce的队列服务从今年四月开始从RabbitMQ切换到了MongoDB上,并一直稳定运行至今,下面是BoxedIce对队列系统的一个讨论PPT及相关叙述。分享给大家。

为什么要使用队列系统?

  • 因为一些任务需要在后台执行,让调用者不需要等待其完成就能返回。比如给用户发送邮件,短信之类的操作。
  • 还有就是一个大系统内部的通信,可能会采用队列的方式传递消息。

对队列系统有哪些要求?

  • 应对任务处理者:通常会有一些进程从队列里获取消息进行处理,而且通常这些进程都会启动很多个。所以队列需要能够处理并发的数据请求操作。
  • 原子性:队列中的元素只能被取出一次,必须保证每次读取队列中元素进行操作和删除这个元素是原子性的。
  • 快速:队列系统要能够快速地处理元素的写入和读取操作。
  • 垃圾回收能力:如果一个任务处理到一半死掉了,那么必须能有方法监测到并且将这个任务重新放入队列中

关于任务处理

任务处理进程的工作,就是从队列中读出消息,并且处理这个消息。所以它需要一种能够从队列中取出元素进行处理的能力。RabbitMQ提供了AMQP协议,目前已经有许多基于此协议的客户端了,相对的,MongoDB基于其Mongo Wire Protocol协议也拥有丰富的客户端支持。

BoxedIce在使用RabbitMQ时是用的pika客户端,转为MongoDB后使用的是pymongo。这两个协议相对比,pymongo的开销要小很多。

对于原子性,RabbitMQ通过对consume/ack协议的支持来实现。而由于MongoDB只支持对单个文档的原子性个性,所以你可以使用其findAndModify 命令,简单语法如下:

db.runCommand( { findAndModify : collection, { options } } )

这里的options是一个数组,其包含下面一些元素:

  • where:是一个查询条件,比如在我们的例子里,这个查询条件是{‘where’ { ‘inProg’: false, ‘done’: false } }操作会针对查询到的第一个元素进行。
  • sort:是对查询结构的排序,可以设置查询返回结果是按哪种条件排序的。比如你可以设置一个消息优先级,然后按优先级的倒序排序,让优先级高的消息优先进行处理。
  • update:这是标明你需要如何个性这个元素,在我们的例子里,我们设置其inProg标识为true,标明正在处理中,设置处理时间t为当前时间,这个时间会用在垃圾回收中。如下:{‘update’ : {‘$set’ : { ‘inProg’ : true, ‘t’ : new Date() } } }

垃圾回收

上面我们在处理消息的时候会将其inProg设置为true表示正在处理,当成功处理完成后,再将done设置为true,但如果处理过程中出现问题,就会导致一个inProg为true但是done永远为false的任务。这时候就需要垃圾回收策略通过检查处理时间t来进行垃圾回收处理了。

now = datetime.datetime.now()
difference = datetime.timedelta(seconds=10)
timeout = now - difference

queue.find({'inProg' : True, 'start' : {'$lte' : timeout} })

比如上面的代码,我们通过判断t是否是在当前时间10秒以前来判断是否过期(10秒都未处理完,我们认为任务处理失败),这些失败的消息我们可以进行相关的处理,让它重新加入到消息队列中。

其它的一些考虑

除了上面说到的速度,原子性等特性,对于一个队列系统,还是有一些其它方面需要考虑的。

  • 容错性:MongoDB的replica sets架构提供了整体的高可用性。当其被用作队列时,也同样继承了这一我。而RabbitMQ并没有内置的支持。目前在RabbitMQ 2.6.0中有相关的支持
  • 一致性:MongoDB默认会一分钟将数据flush到磁盘,但其同时提供一个默认100ms的操作日志可以增强其单机的可靠性。可以缓解宕机时数据丢失导致不一至的情况。如果你对一致性要求非常高,你也可以使用MongoDB的getLastError命令来保证你的每次操作都写入操作日志或者磁盘上才返回成功。
  • 扩展性:我们使用capped collection来做消息队列,所以老数据的清除是自动的。在MongoDB中可以通过sharding方式来实现数据的横向扩展,但是sharding并不支持用于capped collection。你可以自己选择自己需要的应用方式。

来源:blog.boxedice.com