反应式java如何从couchbase访问数据？

import com.couchbase.client.java.Cluster;
import com.couchbase.client.java.Collection;
import com.couchbase.client.java.json.JsonObject;
import com.couchbase.client.java.kv.GetResult;
import com.couchbase.client.java.kv.MutationResult;
import com.couchbase.client.java.kv.UpsertOptions;
import reactor.core.publisher.Mono;

public class ReactiveJavaCouchbaseExample {
    private Cluster cluster;
    private Collection collection;

    public ReactiveJavaCouchbaseExample() {
        // 初始化Couchbase集群和集合
        cluster = Cluster.connect("couchbase://localhost", "username", "password");
        collection = cluster.bucket("bucketName").defaultCollection();
    }

    public Mono<GetResult> getData(String key) {
        // 异步获取数据
        return collection.get(key);
    }

    public Mono<MutationResult> saveData(String key, JsonObject data) {
        // 异步保存数据
        return collection.upsert(key, data, UpsertOptions.upsertOptions().expiry(3600));
    }
}