MongoDb Streaming von eingefügten Daten in Echtzeit (oder nahezu Echtzeit)

Question

Ich habe eine Reihe von MongoDB-Sammlungen, die eine Reihe von JSON-Dokumenten aus verschiedenen Streaming-Quellen aufnehmen. Mit anderen Worten: Es gibt eine Reihe von Prozessen, die kontinuierlich Daten in eine Reihe von MongoDB-Sammlungen einfügen.

Ich brauche eine Möglichkeit, die Daten aus MongoDB in nachgelagerte Anwendungen zu streamen. Ich möchte also ein System, das konzeptionell wie folgt aussieht:

App Stream1 --> 
App Stream2 -->     MONGODB     --->  Aggregated Stream
App Stream3 -->

ODER dies:

App Stream1 -->                 --->  MongoD Stream1
App Stream2 -->     MONGODB     --->  MongoD Stream2
App Stream3 -->                 --->  MongoD Stream3

Die Frage ist, wie kann ich Daten aus Mongo streamen, ohne ständig die Datenbank abfragen zu müssen?

Die offensichtliche Frage Antwort wäre, "warum nicht Sie diese App Streaming-Prozesse ändern, um Nachrichten an eine Warteschlange wie Rabbit, Zero oder ActiveMQ zu senden, die dann hat sie an Ihre Mongo Streaming-Prozesse und Mongo auf einmal wie diese senden":

                 MONGODB
                   /|\  
                    |
App Stream1 -->     |          --->  MongoD Stream1
App Stream2 -->  SomeMQqueue   --->  MongoD Stream2
App Stream3 -->                --->  MongoD Stream3

In einer idealen Welt wäre das gut, aber wir brauchen Mongo, um sicherzustellen, dass Nachrichten zuerst gespeichert werden, um Duplikate zu vermeiden und sicherzustellen, dass alle IDs generiert werden usw. Mongo muss in der Mitte als die Persistenzschicht sitzen.

Wie streame ich also Nachrichten aus einer Mongo-Sammlung (ohne GridFS usw.) in diese Downstream-Anwendungen? Der Grundgedanke war, einfach nach neuen Dokumenten zu suchen und jedes gesammelte Dokument zu aktualisieren, indem ein weiteres Feld zu den in der Datenbank gespeicherten JSON-Dokumenten hinzugefügt wird, ähnlich wie ein Prozess-Flag in einer SQL-Tabelle, das einen verarbeiteten Zeitstempel speichert. D.h. alle 1 Sekunde Abfrage nach Dokumenten, bei denen verarbeitet == null.... add processed = now().... update document.

Gibt es eine sauberere/rechnereffizientere Methode?

Zu Ihrer Information: Dies sind alles Java-Prozesse.

Zum Wohl!