I'm creating a sort of background job queue system with MongoDB as the data store. How can I "listen" for inserts to a MongoDB collection before spawning workers to process the job? Do I need to poll every few seconds to see if there are any changes from last time, or is there a way my script can wait for inserts to occur? This is a PHP project that I am working on, but feel free to answer in Ruby or language agnostic.
-
Change Streams was added in MongoDB 3.6 to address your scenario. docs.mongodb.com/manual/changeStreams Also if you are using MongoDB Atlas you can leverage Stitch Triggers which allow you to execute functions in response to insert/update/deletes/etc. docs.mongodb.com/stitch/triggers/overview No more needing to parse the oplog. – Robert Walters Oct 4 '18 at 12:06
What you are thinking of sounds a lot like triggers. MongoDB does not have any support for triggers, however some people have "rolled their own" using some tricks. The key here is the oplog.
When you run MongoDB in a Replica Set, all of the MongoDB actions are logged to an operations log (known as the oplog). The oplog is basically just a running list of the modifications made to the data. Replicas Sets function by listening to changes on this oplog and then applying the changes locally.
Does this sound familiar?
I cannot detail the whole process here, it is several pages of documentation, but the tools you need are available.
First some write-ups on the oplog
- Brief description
- Layout of the local
collection (which contains the oplog)
You will also want to leverage tailable cursors. These will provide you with a way to listen for changes instead of polling for them. Note that replication uses tailable cursors, so this is a supported feature.
-
1hmm...not exactly what I had in mind. I am only running one instance at this point (no slaves). So maybe a more basic solution? – Andrew Mar 13 '12 at 22:18
-
15You can start the server with the
--replSet
option and it will create / populate theoplog
. Even without the secondary. This is definitely the only way to "listen" to changes in the DB. – Gates VP Mar 13 '12 at 22:36 -
2This is a nice description how to setup oplog for logging changes to DB locally: loosexaml.wordpress.com/2012/09/03/… – johndodo Dec 30 '14 at 14:35
-
I know the question is a bit older, but take a look at rethinkdb.com – Marian Klühspies Dec 10 '15 at 23:58
-
Cooooool! That's really what I want. And I found a library named 'mongo-oplog' on npm. So happy~ – pjincz Dec 11 '16 at 20:14
MongoDB has what is called capped collections
and tailable cursors
that allows MongoDB to push data to the listeners.
A capped collection
is essentially a collection that is a fixed size and only allows insertions. Here's what it would look like to create one:
db.createCollection("messages", { capped: true, size: 100000000 })
MongoDB Tailable cursors (original post by Jonathan H. Wage)
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
PHP
$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);
}
}
Python (by Robert Stewart)
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)
Perl (by Max)
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;
}
}
Additional Resources:
An article talking about tailable cursors in more detail.
PHP, Ruby, Python, and Perl examples of using tailable cursors.
-
62
-
2@rbp haha, I never said it was production code, but you're right, sleeping for a second is not a good practice. Pretty sure I got that example from somewhere else. Not sure how to refactor it though. – Andrew Sep 13 '13 at 16:16
-
13@kroe because those irrelevant details will get put into production code by newer programmers that may not understand why it's bad. – Catfish Jan 14 '15 at 17:35
-
3I understand your point, but expecting some new programmers to add "sleep 1" to production is almost offensive! I mean, i wouldn't be surprised... But if someone puts this in production, at least will learn the hard way and forever.. hahaha – kroe Jan 14 '15 at 18:22
-
14
Since MongoDB 3.6 there will be a new notifications API called Change Streams which you can use for this. See this blog post for an example. Example from it:
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'])
-
3
-
4
-
1how? don't use polling - you need an evented approach instead of while loops, etc. – Alexander Mills Mar 17 '18 at 4:17
-
3
-
I think he/she is referring to the last loop. But I think PyMongo only support that. Motor might have a async/event listener-style implementation. – Shane Hsu Jan 24 at 15:51
Check out this: Change Streams
January 10, 2018 - Release 3.6
*EDIT: I wrote an article about how to do this https://medium.com/riow/mongodb-data-collection-change-85b63d96ff76
https://docs.mongodb.com/v3.6/changeStreams/
It's new in mongodb 3.6 https://docs.mongodb.com/manual/release-notes/3.6/ 2018/01/10
$ mongod --version
db version v3.6.2
In order use changeStreams the database must be a Replication Set
More about Replication Sets: https://docs.mongodb.com/manual/replication/
Your Database will be a "Standalone" by default.
How to Convert a Standalone to a Replica Set: https://docs.mongodb.com/manual/tutorial/convert-standalone-to-replica-set/
The following example is a practical application for how you might use this.
* Specifically for 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 */
Useful links:
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
-
sorry about all edits, SO didn't like my "Links" (said they were improperly formatted code.) – Rio Weber Feb 7 '18 at 18:53
-
1you shouldn't have to query the database, I think with watch() or similar, the new data can be sent to the server that is listening – Alexander Mills Mar 17 '18 at 4:16
MongoDB version 3.6 now includes change streams which is essentially an API on top of the OpLog allowing for trigger/notification-like use cases.
Here is a link to a Java example: http://mongodb.github.io/mongo-java-driver/3.6/driver/tutorials/change-streams/
A NodeJS example might look something like:
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));
});
});
Alternatively, you could use the standard Mongo FindAndUpdate method, and within the callback, fire an EventEmitter event (in Node) when the callback is run.
Any other parts of the application or architecture listening to this event will be notified of the update, and any relevant data sent there also. This is a really simple way to achieve notifications from Mongo.
-
this is very inefficient..you're locking the db for each FindAndUpdate! – Yash Gupta Nov 21 '15 at 16:22
-
1My guess is that Alex was answering a slightly different (not specifically addressing inserts) but related question as in how to fire off some kind of notification to clients when the state of a queued job changes which we assume will need to happen as jobs are spawned, complete successfully or fail. With clients connected using websockets to node, they can all be notified of changes with a broadcast event on the FIndAndUpdate callback which could be called when receive state change messages. I would say that this isn't inefficient as the updates need to be done. – Peter Scott Dec 28 '15 at 21:20
Many of these answers will only give you new records and not updates and/or are extremely ineffecient
The only reliable, performant way to do this is to create a tailable cursor on local db: oplog.rs collection to get ALL changes to MongoDB and do with it what you will. (MongoDB even does this internally more or less to support replication!)
Explanation of what the oplog contains: https://www.compose.com/articles/the-mongodb-oplog-and-node-js/
Example of a Node.js library that provides an API around what is available to be done with the oplog: https://github.com/cayasso/mongo-oplog
There is an working java example which can be found here.
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();
The key is QUERY OPTIONS given here.
Also you can change find query, if you don't need to load all the data every time.
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);
Actually, instead of watching output, why you dont get notice when something new is inserted by using middle-ware that was provided by mongoose schema
You can catch the event of insert a new document and do something after this insertion done
protected by nullpointer Aug 22 '18 at 17:34
Thank you for your interest in this question.
Because it has attracted low-quality or spam answers that had to be removed, posting an answer now requires 10 reputation on this site (the association bonus does not count).
Would you like to answer one of these unanswered questions instead?