原文:http://blog.nosqlfan.com/html/3223.html
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。你可以自己选择自己需要的应用方式。