执行查询时 MongoDB Reactive Streams 挂起

Question

我正在使用我在this example 之后实现的 MongoDB Reactive Streams Java API，但我遇到了一个严重的问题：有时，当我尝试查询集合时，await 方法不起作用，它会一直挂起，直到超时。

onSubscribe 方法被正确调用，但是onNext、onError 和 onComplete 都没有被调用。

似乎没有导致此问题的特定情况。

这是我的代码

MongoDatabase database = MongoDBConnector.getClient().getDatabase("myDb");
MongoCollection<Document> collection = database.getCollection("myCollection");
FindPublisher<Document> finder = collection.find(Filters.exists("myField"));    
SettingSubscriber tagSub = new SettingSubscriber(finder); 
//SettingsSubscriber is a subclass of ObservableSubscriber which calls publisher.subscribe(this)
tagSub.await(); //this is where it hangs
return tagSub.getWrappedData();

Answer 1

我编写了一个我认为SettingSubscriber 看起来像的简单实现，并尝试使用常规脚本重新创建问题。我不能——我的代码运行时没有挂起，打印每个输出记录并退出。以下代码供参考：

@Grab(group = 'org.mongodb', module = 'mongodb-driver-reactivestreams', version = '4.3.3')
@Grab(group = 'org.slf4j', module = 'slf4j-api', version = '1.7.32')
@Grab(group = 'ch.qos.logback', module = 'logback-classic', version = '1.2.6')

import com.mongodb.MongoClientSettings;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import com.mongodb.reactivestreams.client.MongoClients;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoDatabase;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.FindPublisher;
import com.mongodb.client.model.Filters;
import org.bson.Document;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.CountDownLatch;

MongoClientSettings.Builder clientSettingsBuilder = MongoClientSettings.builder()
        .applyToClusterSettings { clusterSettingsBuilder ->
            clusterSettingsBuilder.hosts( Arrays.asList(new ServerAddress("localhost", 27017)))
        };
MongoClient mongoClient = MongoClients.create(clientSettingsBuilder.build());
MongoDatabase database = mongoClient.getDatabase("myDb");
MongoCollection<Document> collection = database.getCollection("myCollection");
FindPublisher<Document> finder = collection.find(Filters.exists("myField"));
SettingSubscriber tagSub = new SettingSubscriber(finder);
tagSub.await();

class SettingSubscriber implements Subscriber<Document> {
    private final CountDownLatch latch = new CountDownLatch(1);
    private Subscription subscription;
    private List<Document> data = new ArrayList<>();

    public SettingSubscriber(FindPublisher<Document> finder) {
        finder.subscribe(this);
    }

    @Override
    public void onSubscribe(final Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(final Document document) {
        System.out.println("Received: " + document);
        data.add(document);
        subscription.request(1);
    }

    @Override
    public void onError(final Throwable throwable) {
        throwable.printStackTrace();
        latch.countDown();
    }

    @Override
    public void onComplete() {
        System.out.println("Completed");
        latch.countDown();
    }
    
    public List<Document> getWrappedData() {
        return data;
    }

    public void await() throws Throwable {
        await(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    }

    public void await(final long timeout, final TimeUnit unit) throws Throwable {
        if (!latch.await(timeout, unit)) {
            System.out.println("Publish timed out");
        }
    }
}

您能否将SettingSubscriber 的此实现与您的进行比较，看看是否遗漏了什么？