使用mongo来构建一个简单的消息队列

MongoDB 有一个叫 Tailable Cursors的特性，它类似于tail -f 命令，你在一个Capped Collection上面执行查询操作，当操作完成后，你可以不关闭返回的数据Cursor，并持续地从中读出新加入的数据。

这个特性可以用来干什么？我觉得最直接的一个用途就是用作消息队列了，利用此特性加上MongoDB 天然的Replication 机制，做一个分布式的队列系统貌似不是什么难事。

Capped collections

Capped collections 就是固定大小的collection。 它有很高的性能以及队列过期的特性(过期按照插入的顺序)。 有点和 "RRD" 概念类似。

What RRDtool does？

RRDtool is the OpenSource industry standard, high performance data logging and graphing system for time series data.

Tailable Cursor

• 通常， mongoDB在遍历完所有结果集中的数据后，会自动关闭游标。
• 但是， 对于Capped collections， 在遍历完所有数据后，游标一直保持打开状态。当客户端再次向Capped collections插入数据时，Tailable Cursor将继续得到数据。

**注：**Tailable Cursor在概念上，类似于Unix的tail命令的-f选项，即一种‘follow’模式。

创建步骤

1. 创建一个 Capped Collection

   db.createCollection("mycoll", {capped:true, size:100000})

   和标准的collection不同，你必须要显式的创建一个capped collection， 指定一个collection的大小，单位是字节。collection的数据存储空间值提前分配的。 要注意的是指定的存储大小包含了数据库的头信息。

   特性 如果空间都被使用完毕，新添加的对象会取代最旧的那个数据。
   • 如果你执行find()，并没有指定顺序。返回的结果就是按照插入顺序排序。
   • 倒序使用 find().sort({$natural:-1})。
2. 应用

   日志Logging.

   Capped collection性能非常优秀，可以来存储日志文档。
   • 插入一个没有索引的capped collection速度非常接近存储在文件系统。 - 由于使用了内置的LRU机制，也不用担心超出硬盘空间。

3. 缓存Caching.

   如果你希望在数据库缓存一些小数量的对象。capped collection提供了非常方便的机制来实现这个操作。
   • 注意的是要给capped table添加索引，因为这种应用，读频率高于写。

4. 自动存档Auto Archiving.

   如果你希望数据自动过期。capped collection要比手写cron scripts更为方便。
5. 建议
   默认的情况下，capped collection不会在_id添加索引。
   • 为了最大化性能，不要再capped collection上创建索引。
   • 如果这个collection写操作多于读操作，更不需要索引了。
   • 注意的是，你可能创建了索引。速度就会降低，但是还是要比标准的collection要快。
   • 使用 natural ordering 来更有效的获取最近插入的元素。和linux的tail命令相似。
6. 限制对象的个数 db.createCollection("mycoll", {capped:true, size:100000, max:100}); 提示: 当编程的时候，存储最近对象的版本号的方法就是把max参数设为1(max=1)。
7. 使用validate()工具来查看collection使用的存储空间 db.mycoll.validate();
8. 查看一个collection是否为capped collection 可以调用isCapped方法来查看一个collection是否为capped collection。 db.foo.isCapped()

9. 使用tailable cursors

   示例： oplog

   注：

   tailable cursors不使用索引
   • 返回文档顺序为硬盘中存储的顺序
   • 因为不用索引，初始查询代价高；新增文档代价低；
   • 游标在以下几种情况会死掉或不可用： 没有返回
   • 返回的文档在集合末尾，并且应用删除了文档
10. 死掉的游标ID为0。

var filter = {};

// set MongoDB cursor options
var cursorOptions = {
tailable: true,
awaitdata: true,
numberOfRetries: -1
};

// create stream and listen
var stream = coll.find(filter, cursorOptions).sort({$natural: -1}).stream();

// call the callback
stream.on('data', function(document) {
console.log(document);
});

mubsub

项目地址： https://github.com/scttnlsn/mubsub

mubsub基于node.js和mongoDB实现发布，订阅消息。

使用mongo的capped collections和tailable cursors，当插入指定文档时，通知订阅者。