Ich erstelle eine Art Hintergrund-Job-Warteschlangensystem mit MongoDB als Datenspeicher. Wie kann ich auf Einfügungen in eine MongoDB-Sammlung "lauschen", bevor ich Arbeiter zur Bearbeitung des Jobs hervorbringe? Muss ich alle paar Sekunden abfragen, um festzustellen, ob sich Änderungen gegenüber dem letzten Mal ergeben haben, oder kann mein Skript auf das Einfügen warten? Dies ist ein PHP-Projekt, an dem ich arbeite, aber ich kann gerne in Ruby oder sprachunabhängig antworten.

Was Sie denken, klingt sehr nach Auslösern. MongoDB hat keine Unterstützung für Trigger, jedoch haben einige Leute mit einigen Tricks "ihre eigenen gewürfelt". Der Schlüssel hier ist das Oplog.

Wenn Sie MongoDB in einem Replikatsatz ausführen, werden alle MongoDB-Aktionen in einem Betriebsprotokoll (bekannt als Oplog) protokolliert. Das Oplog ist im Grunde nur eine laufende Liste der an den Daten vorgenommenen Änderungen. Replikate Legt die Funktion fest, indem Änderungen in diesem Oplog abgehört und die Änderungen dann lokal angewendet werden.

Kommt Ihnen das bekannt vor?

Ich kann den gesamten Prozess hier nicht detaillieren, es sind mehrere Seiten Dokumentation, aber die Tools, die Sie benötigen, sind verfügbar.

Zuerst einige Beschreibungen zum Oplog - Kurzbeschreibung - Layout der localSammlung (die das Oplog enthält)

Sie möchten auch schwanzfähige Cursor nutzen . Diese bieten Ihnen die Möglichkeit, auf Änderungen zu warten, anstatt sie abzufragen. Beachten Sie, dass bei der Replikation anpassbare Cursor verwendet werden. Dies ist also eine unterstützte Funktion.

MongoDB hat das, was so genannt wird, capped collectionsund tailable cursorsdas ermöglicht es MongoDB, Daten an die Listener zu senden.

A capped collectionist im Wesentlichen eine Sammlung mit fester Größe, die nur Einfügungen zulässt. So würde es aussehen, eine zu erstellen:

db.createCollection("messages", { capped: true, size: 100000000 })

MongoDB Tailable Cursor ( Originalbeitrag von Jonathan H. Wage )

Rubin

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 (von 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 (von 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;
    }
}

Zusätzliche Ressourcen:

Ruby / Node.js Tutorial, das Sie durch die Erstellung einer Anwendung führt, die Einfügungen in eine MongoDB-gekappte Sammlung abhört.

Ein Artikel, der ausführlicher über schwanzfähige Cursor spricht.

PHP-, Ruby-, Python- und Perl-Beispiele für die Verwendung von Tailable-Cursorn.

Seit MongoDB 3.6 wird es eine neue Benachrichtigungs-API namens Change Streams geben, die Sie dafür verwenden können. In diesem Blogbeitrag finden Sie ein Beispiel . Beispiel daraus:

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'])

Überprüfen Sie dies: Streams ändern

10. Januar 2018 - Release 3.6

* BEARBEITEN: Ich habe einen Artikel darüber geschrieben, wie das geht https://medium.com/riow/mongodb-data-collection-change-85b63d96ff76

https://docs.mongodb.com/v3.6/changeStreams/

Es ist neu in Mongodb 3.6 https://docs.mongodb.com/manual/release-notes/3.6/ 2018/01/10

$ mongod --version
db version v3.6.2

Um changeStreams verwenden zu können, muss die Datenbank ein Replikationssatz sein

Weitere Informationen zu Replikationssätzen: https://docs.mongodb.com/manual/replication/

Ihre Datenbank ist standardmäßig " Standalone ".

So konvertieren Sie ein eigenständiges in ein Replikatset: https://docs.mongodb.com/manual/tutorial/convert-standalone-to-replica-set/

Das folgende Beispiel ist eine praktische Anwendung, wie Sie dies verwenden können.
* Speziell für 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 */

Nützliche 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

MongoDB Version 3.6 enthält jetzt Änderungsströme, die im Wesentlichen eine API über dem OpLog sind und Trigger- / Benachrichtigungs-ähnliche Anwendungsfälle ermöglichen.

Hier ist ein Link zu einem Java-Beispiel: http://mongodb.github.io/mongo-java-driver/3.6/driver/tutorials/change-streams/

Ein NodeJS-Beispiel könnte ungefähr so ​​aussehen:

 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));
          });
      });

Alternativ können Sie die Standardmethode Mongo FindAndUpdate verwenden und innerhalb des Rückrufs ein EventEmitter-Ereignis (in Node) auslösen, wenn der Rückruf ausgeführt wird.

Alle anderen Teile der Anwendung oder Architektur, die dieses Ereignis abhören, werden über das Update informiert, und alle relevanten Daten werden ebenfalls dorthin gesendet. Dies ist eine sehr einfache Möglichkeit, Benachrichtigungen von Mongo zu erhalten.

Viele dieser Antworten geben Ihnen nur neue Datensätze und keine Aktualisierungen und / oder sind äußerst ineffizient

Die einzige zuverlässige und performante Möglichkeit, dies zu tun, besteht darin, einen anpassbaren Cursor auf der lokalen Sammlung db: oplog.rs zu erstellen, um ALLE Änderungen an MongoDB zu erhalten und damit zu tun, was Sie wollen. (MongoDB tut dies sogar mehr oder weniger intern, um die Replikation zu unterstützen!)

Erklärung, was das Oplog enthält: https://www.compose.com/articles/the-mongodb-oplog-and-node-js/

Beispiel für eine Node.js-Bibliothek, die eine API für die verfügbaren Funktionen des Oplogs bereitstellt: https://github.com/cayasso/mongo-oplog

Es gibt eine fantastische Reihe von Diensten namens MongoDB Stitch . Schauen Sie sich Stichfunktionen / Trigger an . Beachten Sie, dass dies ein Cloud-basierter kostenpflichtiger Dienst (AWS) ist. In Ihrem Fall können Sie auf einer Einfügung eine benutzerdefinierte Funktion aufrufen, die in Javascript geschrieben ist.

Es gibt ein funktionierendes Java-Beispiel, das hier zu finden ist .

 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();

Der Schlüssel sind die hier angegebenen Abfrageoptionen .

Sie können auch die Suchabfrage ändern, wenn Sie nicht jedes Mal alle Daten laden müssen.

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);

Anstatt die Ausgabe zu sehen, werden Sie nicht bemerkt, wenn etwas Neues eingefügt wird, indem Sie Middleware verwenden, die vom Mungo-Schema bereitgestellt wurde

Sie können den Fall des Einfügens eines neuen Dokuments abfangen und nach dem Einfügen etwas unternehmen

Nach 3.6 darf man Datenbank die folgenden Datenbank-Trigger-Typen verwenden:

• ereignisgesteuerte Trigger - nützlich, um verwandte Dokumente automatisch zu aktualisieren, nachgeschaltete Dienste zu benachrichtigen, Daten zu verbreiten, um gemischte Workloads, Datenintegrität und Überwachung zu unterstützen
• Geplante Trigger - nützlich für geplante Workloads zum Abrufen, Weitergeben, Archivieren und Analysieren von Daten

Melden Sie sich bei Ihrem Atlas-Konto an, wählen Sie die TriggersBenutzeroberfläche aus und fügen Sie einen neuen Auslöser hinzu:

Erweitern Sie jeden Abschnitt für weitere Einstellungen oder Details.