我正在使用MongoDB作为数据存储,创建一种后台工作队列系统。如何在生成工作程序以处理任务之前“监听”MongoDB集合的插入?
我需要每隔几秒钟轮询以查看上次是否有任何更改,还是有办法让我的脚本等待插入发生?
这是我正在进行的PHP项目,但请用Ruby或语言不可知的方式回答。
我正在使用MongoDB作为数据存储,创建一种后台工作队列系统。如何在生成工作程序以处理任务之前“监听”MongoDB集合的插入?
我需要每隔几秒钟轮询以查看上次是否有任何更改,还是有办法让我的脚本等待插入发生?
这是我正在进行的PHP项目,但请用Ruby或语言不可知的方式回答。
local
集合的布局(其中包含oplog)--replSet
选项启动服务器,它会创建/填充oplog
,即使没有第二个服务器也可以。这绝对是“监听”数据库更改的唯一方法。 - Gates VPcapped collections
和tailable cursors
,它们允许MongoDB向监听者推送数据。capped collection
本质上是一个固定大小且只允许插入的集合。以下是创建它的示例代码:db.createCollection("messages", { capped: true, size: 100000000 })
Ruby
coll = db.collection('my_collection')
cursor = Mongo::Cursor.new(coll, :tailable => true)
loop do
if doc = cursor.next_document
puts doc
else
sleep 1
end
end
$mongo = new Mongo();
$db = $mongo->selectDB('my_db')
$coll = $db->selectCollection('my_collection');
$cursor = $coll->find()->tailable(true);
while (true) {
if ($cursor->hasNext()) {
$doc = $cursor->getNext();
print_r($doc);
} else {
sleep(1);
}
}
from pymongo import Connection
import time
db = Connection().my_db
coll = db.my_collection
cursor = coll.find(tailable=True)
while cursor.alive:
try:
doc = cursor.next()
print doc
except StopIteration:
time.sleep(1)
use 5.010;
use strict;
use warnings;
use MongoDB;
my $db = MongoDB::Connection->new;
my $coll = $db->my_db->my_collection;
my $cursor = $coll->find->tailable(1);
for (;;)
{
if (defined(my $doc = $cursor->next))
{
say $doc;
}
else
{
sleep 1;
}
}
*编辑:我撰写了一篇关于如何做到这一点的文章https://medium.com/riow/mongodb-data-collection-change-85b63d96ff76
在mongodb 3.6中是新功能, https://docs.mongodb.com/manual/release-notes/3.6/ 2018/01/10
$ mongod --version
db version v3.6.2
为了使用 changeStreams,数据库必须是一个副本集。
有关副本集的更多信息: https://docs.mongodb.com/manual/replication/
默认情况下,您的数据库将是一个"独立"的。
如何将独立数据库转换为副本集:https://docs.mongodb.com/manual/tutorial/convert-standalone-to-replica-set/
以下示例是如何实际应用的。
* 特别适用于 Node。
/* file.js */
'use strict'
module.exports = function (
app,
io,
User // Collection Name
) {
// SET WATCH ON COLLECTION
const changeStream = User.watch();
// Socket Connection
io.on('connection', function (socket) {
console.log('Connection!');
// USERS - Change
changeStream.on('change', function(change) {
console.log('COLLECTION CHANGED');
User.find({}, (err, data) => {
if (err) throw err;
if (data) {
// RESEND ALL USERS
socket.emit('users', data);
}
});
});
});
};
/* END - file.js */
有用的链接:
https://docs.mongodb.com/manual/tutorial/convert-standalone-to-replica-set
https://docs.mongodb.com/manual/tutorial/change-streams-example
https://docs.mongodb.com/v3.6/tutorial/change-streams-example
http://plusnconsulting.com/post/MongoDB-Change-Streams
自 MongoDB 3.6 开始,将会有一个名为 Change Streams 的新通知 API,您可以用它来实现此功能。参见 这篇博客文章的示例。下面是来自该博客文章的一个示例:
cursor = client.my_db.my_collection.changes([
{'$match': {
'operationType': {'$in': ['insert', 'replace']}
}},
{'$match': {
'newDocument.n': {'$gte': 1}
}}
])
# Loops forever.
for change in cursor:
print(change['newDocument'])
MongoDB版本3.6现在包括变更流,这本质上是OpLog之上的API,可以用于触发器/通知等用例。
这里有一个Java示例链接:http://mongodb.github.io/mongo-java-driver/3.6/driver/tutorials/change-streams/
NodeJS示例可能会类似于:
var MongoClient = require('mongodb').MongoClient;
MongoClient.connect("mongodb://localhost:22000/MyStore?readConcern=majority")
.then(function(client){
let db = client.db('MyStore')
let change_streams = db.collection('products').watch()
change_streams.on('change', function(change){
console.log(JSON.stringify(change));
});
});
或者,您可以使用标准的Mongo FindAndUpdate方法,在回调函数中运行EventEmitter事件(在Node中),当回调函数运行时触发该事件。
应用程序或架构的任何其他部分监听此事件将收到更新通知,并发送到那里的任何相关数据也会被通知。这是实现来自Mongo的通知的一种非常简单的方法。
有一组非常棒的服务可供使用,称为MongoDB Stitch。请查看Stitch函数/触发器。请注意,这是一个基于云的付费服务(AWS)。在您的情况下,您可以在插入时调用用JavaScript编写的自定义函数。
这里有一个可用的Java示例,可以在这里找到。
MongoClient mongoClient = new MongoClient();
DBCollection coll = mongoClient.getDatabase("local").getCollection("oplog.rs");
DBCursor cur = coll.find().sort(BasicDBObjectBuilder.start("$natural", 1).get())
.addOption(Bytes.QUERYOPTION_TAILABLE | Bytes.QUERYOPTION_AWAITDATA);
System.out.println("== open cursor ==");
Runnable task = () -> {
System.out.println("\tWaiting for events");
while (cur.hasNext()) {
DBObject obj = cur.next();
System.out.println( obj );
}
};
new Thread(task).start();
这里提供的关键是查询选项。
如果您不需要每次加载所有数据,还可以更改查找查询。
BasicDBObject query= new BasicDBObject();
query.put("ts", new BasicDBObject("$gt", new BsonTimestamp(1471952088, 1))); //timestamp is within some range
query.put("op", "i"); //Only insert operation
DBCursor cur = coll.find(query).sort(BasicDBObjectBuilder.start("$natural", 1).get())
.addOption(Bytes.QUERYOPTION_TAILABLE | Bytes.QUERYOPTION_AWAITDATA);