java - Spring boot MongoDB Change Streams 不起作用

Question

我尝试从 MongoDB 订阅 watch，但不起作用。这是我的订阅代码:

CodecRegistry pojoCodecRegistry = org.bson.codecs.configuration.CodecRegistries.
            fromRegistries(MongoClientSettings.getDefaultCodecRegistry(), org.bson.codecs.configuration.CodecRegistries.fromProviders(PojoCodecProvider
                    .builder()
                    .automatic(true)
                    .build()));
    MongoDatabase db = mongoClient.getDatabase("test").withCodecRegistry(pojoCodecRegistry);
    MongoCollection<Product> collection = db.getCollection("product", Product.class);
    collection.watch(Product.class)
            .subscribe(new Subscriber<ChangeStreamDocument<Product>>() {
                @Override
                public void onSubscribe(Subscription subscription) {
                    System.out.println("onSubscribe");
                }

                @Override
                public void onNext(ChangeStreamDocument<Product> documentChangeStreamDocument) {
                    System.out.println("onNext");

                }

                @Override
                public void onError(Throwable throwable) {
                    System.out.println("onError");

                }

                @Override
                public void onComplete() {
                    System.out.println("onComplete");

                }
            });

我只能在控制台上看到此文本“onSubscribe”，但它不适用于 onNext 函数。我不知道我做错了什么。

感谢您的建议和帮助。

Answer 1

我也遇到了同样的问题。你的 mongoDB 版本是什么我将我的MongoDB版本更新到V4.2添加副本集，然后就可以了

 MongoClient mongoClient = new MongoClient("serverIp", port);
 MongoDatabase db = mongoClient.getDatabase("databaseName");
 MongoCollection<Document> collection = db.getCollection("collectionName");
 List<Bson> pipeline = singletonList(Aggregates.match(
       Filters.in("operationType", asList("insert", "update", "delete"))));
 MongoCursor<ChangeStreamDocument<Document>> watchCursor = 
 collection.watch().iterator();
 LOGGER.info(String.valueOf(watchCursor));
 LOGGER.info("== open cursor ==");

 Runnable task = () -> {
     LOGGER.info("\tWaiting for events");
     while (watchCursor.hasNext()) {
         LOGGER.info("Yes!!!");
         LOGGER.info(String.valueOf(watchCursor.next()));
     }
 };
 new Thread(task).start();