在Couchbase中的异步批量操作

codingman included in NoSQL Persistence Spring Java

2018-06-28 1070 words 3 minutes

Contents

1. 简介

在我们关于在 Spring 应用程序中使用Couchbase 的教程的后续部分中，我们探讨了 Couchbase SDK 的异步特性以及如何使用它来批量执行持久性操作，从而使我们的应用程序能够实现 Couchbase 资源的最佳使用。

1.1. CrudService接口

首先，我们扩充我们的通用CrudService接口以包含批处理操作：

public interface CrudService<T> {
    ...

    List<T> readBulk(Iterable<String> ids);
    void createBulk(Iterable<T> items);
    void updateBulk(Iterable<T> items);
    void deleteBulk(Iterable<String> ids);
    boolean exists(String id);
}

1.2. CouchbaseEntity接口

我们为要持久化的实体定义一个接口：

public interface CouchbaseEntity {
    String getId();
    
    void setId(String id);
    
}

1.3. AbstractCrudService类

然后我们将在一个通用抽象类中实现这些方法中的每一个。该类派生自我们在上一教程中使用的 PersonCrudService类，开头如下：

public abstract class AbstractCrudService<T extends CouchbaseEntity> implements CrudService<T> {
    private BucketService bucketService;
    private Bucket bucket;
    private JsonDocumentConverter<T> converter;
    public AbstractCrudService(BucketService bucketService, JsonDocumentConverter<T> converter) {
        this.bucketService = bucketService;
        this.converter = converter;
    }
    protected void loadBucket() {
        bucket = bucketService.getBucket();
    }
    
    ...
}

2. AsyncBucket接口

Couchbase SDK 提供了用于执行异步操作的AsyncBucket接口。给定一个Bucket实例，你可以通过*async()*方法获取它的异步版本：

AsyncBucket asyncBucket = bucket.async();

3. 批量操作

为了使用AsyncBucket接口执行批处理操作，我们使用了*RxJava *库。

3.1. 批量读取

这里我们实现了readBulk方法。首先我们使用 RxJava 中的AsyncBucket和flatMap机制将文档异步检索到Observable中，然后我们使用 RxJava 中的toBlocking机制将它们转换为实体列表：

@Override
public List<T> readBulk(Iterable<String> ids) {
    AsyncBucket asyncBucket = bucket.async();
    Observable<JsonDocument> asyncOperation = Observable
      .from(ids)
      .flatMap(new Func1<String, Observable<JsonDocument>>() {
          public Observable<JsonDocument> call(String key) {
              return asyncBucket.get(key);
          }
    });
    List<T> items = new ArrayList<T>();
    try {
        asyncOperation.toBlocking()
          .forEach(new Action1<JsonDocument>() {
              public void call(JsonDocument doc) {
                  T item = converter.fromDocument(doc);
                  items.add(item);
              }
        });
    } catch (Exception e) {
        logger.error("Error during bulk get", e);
    }
    return items;
}

3.2. 批量插入

我们再次使用RxJava 的 flatMap构造来实现createBulk方法。

由于批量突变请求的生成速度比生成响应的速度要快，有时会导致过载情况，因此每当遇到BackpressureException时，我们都会以指数延迟重试：

@Override
public void createBulk(Iterable<T> items) {
    AsyncBucket asyncBucket = bucket.async();
    Observable
      .from(items)
      .flatMap(new Func1<T, Observable<JsonDocument>>() {
          @SuppressWarnings("unchecked")
          @Override
          public Observable<JsonDocument> call(final T t) {
              if(t.getId() == null) {
                  t.setId(UUID.randomUUID().toString());
              }
              JsonDocument doc = converter.toDocument(t);
              return asyncBucket.insert(doc)
                .retryWhen(RetryBuilder
                  .anyOf(BackpressureException.class)
                  .delay(Delay.exponential(TimeUnit.MILLISECONDS, 100))
                  .max(10)
                  .build());
          }
      })
      .last()
      .toBlocking()
      .single();
}

3.3. 批量更新

我们在updateBulk方法中使用了类似的机制：

@Override
public void updateBulk(Iterable<T> items) {
    AsyncBucket asyncBucket = bucket.async();
    Observable
      .from(items)
      .flatMap(new Func1<T, Observable<JsonDocument>>() {
          @SuppressWarnings("unchecked")
          @Override
          public Observable<JsonDocument> call(final T t) {
              JsonDocument doc = converter.toDocument(t);
              return asyncBucket.upsert(doc)
                .retryWhen(RetryBuilder
                  .anyOf(BackpressureException.class)
                  .delay(Delay.exponential(TimeUnit.MILLISECONDS, 100))
                  .max(10)
                  .build());
          }
      })
      .last()
      .toBlocking()
      .single();
}

3.4. 批量删除

我们编写deleteBulk方法如下：

@Override
public void deleteBulk(Iterable<String> ids) {
    AsyncBucket asyncBucket = bucket.async();
    Observable
      .from(ids)
      .flatMap(new Func1<String, Observable<JsonDocument>>() {
          @SuppressWarnings("unchecked")
          @Override
          public Observable<JsonDocument> call(String key) {
              return asyncBucket.remove(key)
                .retryWhen(RetryBuilder
                  .anyOf(BackpressureException.class)
                  .delay(Delay.exponential(TimeUnit.MILLISECONDS, 100))
                  .max(10)
                  .build());
          }
      })
      .last()
      .toBlocking()
      .single();
}

4. PersonCrudService

最后，我们编写了一个 Spring 服务PersonCrudService，它为Person实体扩展了AbstractCrudService。

由于所有的 Couchbase 交互都在抽象类中实现，实体类的实现很简单，因为我们只需要确保注入所有依赖项并加载桶：

@Service
public class PersonCrudService extends AbstractCrudService<Person> {
    @Autowired
    public PersonCrudService(
      @Qualifier("TutorialBucketService") BucketService bucketService,
      PersonDocumentConverter converter) {
        super(bucketService, converter);
    }
    @PostConstruct
    private void init() {
        loadBucket();
    }
}