Couchbaseでの非同期バッチ操作

投稿日: 2019-10-19 2022-11-01
タグ: Couchbase, NoSQL, Persistence, Spring

1. 序章

SpringアプリケーションでのCouchbaseの使用に関するチュートリアルのフォローアップでは、Couchbase SDKの非同期性と、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. 非同期バケットインターフェイス

Couchbase SDKは、非同期操作を実行するためのAsyncBucketインターフェイスを提供します。 Bucket インスタンスを指定すると、 async（）メソッドを介して非同期バージョンを取得できます。

AsyncBucket asyncBucket = bucket.async();

3. バッチ操作

AsyncBucket インターフェースを使用してバッチ操作を実行するには、RxJavaライブラリを使用します。

3.1. バッチ読み取り

ここでは、readBulkメソッドを実装します。まず、 AsyncBucket そしてその flatMap ドキュメントを非同期で取得するRxJavaのメカニズム観察可能 、次に使用します toBlocking のメカニズム RxJava これらをエンティティのリストに変換するには：

@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

最後に、PersonエンティティのAbstractCrudServiceを拡張するSpringサービスPersonCrudServiceを記述します。

すべての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();
    }
}

5. 結論

このチュートリアルに示されているソースコードは、githubプロジェクトで入手できます。

Couchbase Java SDKの詳細については、公式のCouchbase開発者向けドキュメントサイトを参照してください。

getdocs

13036